小桥流水78

这个屌丝很懒，什么也没留下！

热门标签

JUC（java.util.concurrent）的常见类

作者：小桥流水78 | 2024-08-21 13:11:08

踩

前言

在前面我们已经学习java.util.concurrent包（实现多线程并发编程常用的包）中的一些操作，例如线程池、阻塞队列等，那么本篇我们就来学习一下JUC中几种其他常见的类。

Callable和Future

Callable

Callable是一个泛型接口，只有一个方法 call().用于创建可以返回结果的任务，与Runnable不同，Callable可以返回一个结果，并且可以抛出异常，需要依赖FutureTask类来获取返回结果。

我们查看Callable的接口定义：

在使用Callable接口的时候，我们需要将我们的任务需求编写到call()方法中。

示例：现在要计算从1加到1000的和


class Demo1 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Callable<Integer> call = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int sum = 0;
                for (int i = 0; i <= 1000; i++) {
                    sum += i;
                }
                return sum;
            }
        };
    }
}

在这个代码中，我们实现了从1加到1000的和的计算，当在完成计算之后，会返回一个Integer类型的数据。

Future

Future同样是一个泛型接口，用来表示异步计算的结果，在接口中提供了一些方法来检查任务是否完成、获取计算结果以及取消任务的执行。以下是Future接口的定义：


public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

cancel((boolean mayInterruptIfRunning):尝试取消任务的执行；
isCancelled()：检查任务是否已经被取消；
isDone()：检查任务是否已经完成；
get()：获取任务的结果，如果任务尚未完成，就会阻塞当前线程，直到任务完成；
get(long timeout, TimeUnit unit)：在指定的时间内获取任务的结果，如果任务尚未完成，则阻塞当前线程。

Future有它的实现类FutureTask供我们使用，我们可以看其文档解释，这段话的意思就是说：可取消的异步计算。FutureTask类提供了Future的基本实现，其中包括启动和取消计算、查询计算是否完成以及检索计算结果的方法。只有在计算完成后才能检索结果;如果计算尚未完成，get方法将阻塞。一旦计算完成，就不能重新启动或取消计算(除非使用runAndReset调用计算)。
FutureTask可以用来包装一个Callable或Runnable对象。因为FutureTask实现了Runnable，所以FutureTask可以提交给Executor执行。
除了作为一个独立的类，这个类还提供了受保护的功能，这在创建自定义任务类时可能很有用。

我们可以在java8文档中查看相关方法：

使用Callable和FutureTask

示例一：创建线程计算1+2+...+1000的和，使用Callable.


    /**
     * 主函数，用于演示如何通过Callable接口实现多线程计算
     * 本函数的目标是计算1到1000的和，通过创建一个Callable任务，并在新线程中执行该任务来实现
     * @param args 命令行参数
     * @throws ExecutionException 如果任务未能成功完成，则抛出此异常
     * @throws InterruptedException 如果当前线程在等待任务完成时被中断，则抛出此异常
     */
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建一个Callable对象，用于计算1到1000的和
        Callable<Integer> call=new Callable<Integer>() {
            /**
             * 执行计算任务的方法
             * 本方法通过循环计算1到1000的和
             * @return 计算结果，即1到1000的和
             * @throws Exception 如果在计算过程中发生错误，则抛出此异常
             */
            @Override
            public Integer call() throws Exception {
                int sum=0;
                for(int i=0;i<=1000;i++){
                    sum+=i;
                }
                return sum;
            }
        };
 
        // 使用Callable对象创建一个FutureTask，以便可以在新线程中执行任务
        FutureTask<Integer> futureTask=new FutureTask<>(call);
 
        // 创建一个新线程，并将FutureTask对象作为任务传递给该线程
        Thread t=new Thread(futureTask);
        // 启动新线程，开始执行计算任务
        t.start();
        // 获取计算任务的结果，主线程将在这里等待直到任务完成
        int result=futureTask.get();
        // 打印计算结果
        System.out.println(result);
    }

结果

示例二：使用单个线程的线程池


    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建一个固定大小的线程池，这里只创建一个线程
        ExecutorService ex=Executors.newFixedThreadPool(1);
        // 创建一个Callable对象，用于执行计算任务
        Callable<Integer> calls=new Callable<Integer>() {
 
            @Override
            public Integer call() throws Exception {
                // 初始化求和变量
                int sum=0;
                // 计算从0到1000的累加和
                for(int i=0;i<=1000;i++){
                    sum+=i;
                }
                // 返回计算结果
                return sum;
            }
        };
        // 创建一个FutureTask对象，并将Callable对象作为参数传递给它
        FutureTask<Integer> futureTask= (FutureTask<Integer>) ex.submit(calls);
        // 获取计算任务的结果，主线程将在这里等待直到任务完成
        int result=futureTask.get();
        // 打印计算结果
        System.out.println(result);
        
        // 关闭ExecutorService
        ex.shutdown();
    }

在这段代码中，我们创建了一个只有一个线程的线程池，并且创建了一个Callable对象并重写其中的call方法。通过调用submit方法提交任务并获取FutureTask对象。但由于此时返回的是Futue类型的结果，所以我们需要将其强转为FutureTask<Integer>。同时我们通过调用其中的get方法来等待任务执行完成并获取到结果。需要注意，这里我们需要手动关闭线程池。释放资源。

示例三：拥有多个线程的线程池


class Demo4{
    // 主函数，展示如何使用固定大小的线程池执行Callable任务
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建一个固定大小为4的线程池
        ExecutorService ex=Executors.newFixedThreadPool(4);
        // 创建一个可返回结果的Callable任务
        Callable<String> calls=new Callable<String>() {
            // 实现call方法，当任务被线程池执行时，此方法将被调用
            @Override
            public String call() throws Exception {
                // 输出当前线程的名称和执行的信息
                System.out.println(Thread.currentThread().getName()+"执行了call方法");
                // 返回任务执行结果
                return "result";
            }
        };
        // 循环提交4个任务到线程池
        for(int i=0;i<4;i++){
            // 提交任务并获取FutureTask对象，用于查询任务执行结果
            FutureTask<String> futureTask= (FutureTask<String>) ex.submit(calls);
            // 获取并输出任务的执行结果
            System.out.println(futureTask.get());
        }
        // 关闭线程池，不再接受新的任务
        ex.shutdown();
    }
}

ReentrantLock

ReentrantLock是一个可重入互斥锁，允许一个线程多次获取同一个锁而不会产生死锁，和synchronized类似，都是用来保证线程安全的。

ReentrantLock加锁有两种方式：

lock()：加锁，如果获取不到锁就进入阻塞等待。
trylock()：加锁，如果获取不到锁，就会放弃加锁。
unlock()：解锁。

在使用ReentrantLock的时候需要手动释放锁unlock(),如果忘记解锁，可能会带来比较严重的后果。所以我们可以使用try-finally来进行结果操作。


        ReentrantLock lock=new ReentrantLock();
        try{
            lock.lock();
            //working
        }finally {
            lock.unlock();
        }

ReentrantLock和synchronized的区别

ReentrantLock可以使用lock()和tryLock()进行加锁，使用lock()的时候，若没有获取到锁就会进入阻塞等待，而使用tryLock()的时候，如果没有获取到锁，就会放弃获取而不是阻塞等待。
synchronized是一个关键字，是JVM内部实现的；ReentrantLock是标准库中的一个类，在JVM外部实现的（基于Java实现）。
synchronized不需要手动解锁，而ReentrantLock需要手动释放锁，使用起来更加灵活，但是也容易遗漏unlock，所以最好在加上try-finally使用。
synchronized在申请锁失败时,会死等.ReentrantLock可以通过trylock的⽅式等待⼀段时间就放弃.
synchronized是⾮公平锁,ReentrantLock默认是⾮公平锁.可以通过构造⽅法传⼊⼀个true开启公平锁模式.
更强⼤的唤醒机制.synchronized是通过Object的wait/notify实现等待-唤醒.每次唤醒的是⼀个随机等待的线程.而ReentrantLock搭配Condition类实现等待-唤醒,可以更精确控制唤醒某个指定的线程.

如何选择使用哪个锁？

锁竞争不激烈的时候,使⽤synchronized,效率更⾼,⾃动释放更⽅便
锁竞争激烈的时候,使⽤ReentrantLock,搭配trylock更灵活控制加锁的⾏为,⽽不是死等.
如果需要使⽤公平锁,使⽤ReentrantLock.

虽然ReentrantLock使用起来比较灵活，但一般情况下建议使用synchronized。

原子类

原子类内部其实是使用CAS实现的，CAS是一个原子的操作，不需要加锁，性能比加锁的要好多。

• AtomicBoolean

• AtomicInteger

• AtomicIntegerArray

• AtomicLong

• AtomicReference

• AtomicStampedReference

CAS的原理以及如何使用CAS在上一篇我已经讲过，想了解更多的可以看看CAS原理

线程池

线程池是为了解决频繁创建和销毁线程带来的性能和资源浪费问题。

在java中使用线程池我们需要用到ExecutorService和Executors。

ExecutorService 表示一个线程池实例.
Executors 是一个工厂类, 能够创建出几种不同风格的线程池.
ExecutorService 的 submit 方法能够向线程池中提交若干个任务

Executors是一个工厂类，里面提供了创建不同风格线程池的方法。

newFixedThreadPool: 创建固定线程数的线程池
newCachedThreadPool: 创建线程数目动态增长的线程池.
newSingleThreadExecutor: 创建只包含单个线程的线程池.
newScheduledThreadPool: 设定延迟时间后执行命令，或者定期执行命令. 是进阶版的 Timer。

Executors本质上是对ThreradPoolExecutor的封装。

感兴趣的可以看看这一篇【JavaEE】线程池-CSDN博客，里面讲解了关于ThreadPoolExecutor构造方法的相关参数的含义。

示例：


class Demo5{
    /**
     * 程序入口
     * @param args 命令行参数
     */
    public static void main(String[] args) {
        // 创建一个固定大小的线程池，用于执行任务
        ExecutorService ex=Executors.newFixedThreadPool(4);
        
        // 提交一个运行时打印"hello"的无返回值任务到线程池
        ex.submit(()->{
            System.out.println("hello");
        });
        
        // 关闭线程池，不再接受新的任务提交，但已提交的任务将继续执行完成
        ex.shutdown();
    }
 
}

Semaphore（信号量）

信号量（Semaphore）又称为信号灯。用来表示“可用资源的个数”，本质上就是一个计数器。在多线程环境下用于协调各个线程, 以保证它们能够正确、合理的使用公共资源。

信号量是一个非负整数，当我们申请资源的时候，计数器就会-1，称为“P操作”，释放资源的时候，计数器就会+1，也称为“V操作”。当计数器为0时，如果再申请资源，就会阻塞等待，直到有其他线释放资源。

Semaphore的PV操作中的加减计数器操作都是原⼦的,可以在多线程环境下直接使用.

信号量可以分为二元信号量和计数信号量两种，二元信号量就相当于一把锁，当有线程申请资源时，就相当于加锁，此时计数器-1，信号量就为0；当有其他线程想要申请资源就会陷入阻塞等待，直到占用资源的线程释放，才能获取到。计数信号量就表示此时有多少可以被申请的资源，当有线程申请资源，计数器就-1，当有线程释放资源，计数器就+1。

加锁和解锁的操作就可以看做，加锁时信号量为0，解锁信号量为1.

在java中，把信号量的相关操作封装在Semaphore中，acquire()方法表示申请资源，release()表示释放资源。availablePermits()可以查看信号量中还有多少资源可用。

示例：


/**
 * SemaphoreDemo类用于演示信号量的应用
 */
class SemaphoreDemo{
    // 用于计数的共享变量
    static int count=0;
    /**
     * 程序的入口点
     * @param args 命令行参数
     * @throws InterruptedException 如果线程被中断
     */
    public static void main(String[] args) throws InterruptedException {
        // 创建一个信号量，初始值为1，用于控制同时只能有一个线程执行临界区代码
        Semaphore semaphore=new Semaphore(1);
        // 创建一个公平的可重入锁，本例中未直接使用，但展示了如何创建
        ReentrantLock locker=new ReentrantLock(true);
        // 创建第一个线程，用于增加计数器
        Thread t1=new Thread(()->{
            for (int i=0;i<1000;i++) {
                try {
                    // 获取信号量，允许进入临界区
                    semaphore.acquire();
                    // 执行临界区操作：增加计数器
                    count++;
                    // 释放信号量，允许其他线程进入临界区
                    semaphore.release();
                } catch (InterruptedException e) {
                    // 如果线程被中断，则抛出运行时异常
                    throw new RuntimeException(e);
                }
            }
        });
        // 创建第二个线程，同样用于增加计数器
        Thread t2=new Thread(()->{
            for (int i=0;i<1000;i++) {
                try {
                    // 获取信号量，允许进入临界区
                    semaphore.acquire();
                    // 执行临界区操作：增加计数器
                    count++;
                    // 释放信号量，允许其他线程进入临界区
                    semaphore.release();
                } catch (InterruptedException e) {
                    // 如果线程被中断，则抛出运行时异常
                    throw new RuntimeException(e);
                }
            }
        });
        // 启动第一个线程
        t1.start();
        // 启动第二个线程
        t2.start();
        // 等待第一个线程结束
        t1.join();
        // 等待第二个线程结束
        t2.join();
        // 输出计数结果
        System.out.println(count);
    }
}

CountDownLatch

CountDownLatch是java中的一个同步工具类，用来协调多个线程之间的同步，初始值为线程的数量。

CountDownLatch通过一个计数器功能，当一个线程完成了自己的任务，计数器的值就-1，当计数器为0时，说明所有的线程都完成任务，此时在CountDownLatch等待的线程就可以恢复执行。

CountDownLatch一个典型用法就是把一个大任务拆分成N个小任务，让多个线程来执行小任务，每个线程执行完自己的任务计数器就-1，当所有小任务都完成后，等待所有小任务完成的线程才继续往下执行。就好比富士康手机加工的流水线一样，组装一步手机需要一条条的流水线来相互配合完成。一条条流水线（Worker），每条线都干自己的活。有的流水线是贴膜的，有的流水线是打螺丝的，有的流水线是质检的、有的流水线充电的、有的流水线贴膜的。等这些流水线都干完了就把一部手机组装完成了。

方法：

CountDownLatch(int count)：count为计数器的初始值（一般需要多少个线程执行，count就设为几）。
countDown(): 每调用一次计数器值-1，直到count被减为0，代表所有线程全部执行完毕。
getCount()：获取当前计数器的值。
await(): 等待计数器变为0，即等待所有异步线程执行完毕。
boolean await(long timeout, TimeUnit unit)：

此方法至多会等待指定的时间，超时后会自动唤醒，若 timeout 小于等于零，则不会等待
boolean 类型返回值：若计数器变为零了，则返回 true；若指定的等待时间过去了，则返回 false


class CountDownLatchDemo{
    public static void main(String[] args) throws InterruptedException {
        //创建一个固定大小的线程池，用于处理下载任务
        ExecutorService service= Executors.newFixedThreadPool(4);
        //创建一个计数器，用于等待所有下载任务完成
        CountDownLatch count=new CountDownLatch(20);//20个任务
        //提交20个下载任务到线程池
        for(int i=1;i<=20;i++){
            int id=i;
            //使用lambda表达式创建匿名内部类，定义每个下载任务的执行逻辑
            service.submit(()->{
                System.out.println("下载任务"+id+"正在执行");
                try {
                    //模拟下载任务的执行时间
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    //处理中断异常
                    throw new RuntimeException(e);
                }
                //任务完成
                System.out.println("下载任务"+id+"执行完毕");
                //计数器减1，表示一个任务完成
                count.countDown();
            });
        }
        //等待所有任务完成，计数器归零
        count.await();
        System.out.println("所有下载任务已完成");
        //关闭线程池
        service.shutdown();
    }
}

当调用了20次 countDown() 方法之后，await() 方法才会结束等待，继续执行后面的代码。

需要注意的是，CountDownLatch是一次性的，一旦计数器的值达到0，就不能再次使用。如果需要多次使用类似的功能，可以考虑使用CyclicBarrier等其他同步工具类。

JUC（java.util.concurrent）的常见类

目录

前言

Callable和Future

Callable

Future

使用Callable和FutureTask

ReentrantLock

ReentrantLock和synchronized的区别

如何选择使用哪个锁？

原子类

线程池

Semaphore（信号量）

CountDownLatch

相关面试题

1.线程同步的方式有哪些？

2.为什么有了synchronized还需要JUC下的lock？

3.AtomicInteger 的实现原理是什么？

4) 信号量听说过么？之前都用在过哪些场景下？

5. 解释⼀下ThreadPoolExecutor构造⽅法的参数的含义