AQS同步组件-CountDownLatch解析和案例

Posted 共饮一杯无

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了AQS同步组件-CountDownLatch解析和案例相关的知识,希望对你有一定的参考价值。

CountDownLatch原理

CountDownLatch是通过一个计数器来实现的,计数器的初始化值为线程的数量。每当一个线程完成了自己的任务后,计数器的值就相应得减1。当计数器到达0时,表示所有的线程都已完成任务,然后在闭锁上等待的线程就可以恢复执行任务。CountDownLatch可以起到阻塞线程,并保证线程在满足某种特定的情况后继续执行。

CountDownLatch 的两种典型用法

  • 某一线程在开始运行前等待n个线程执行完毕。将 CountDownLatch 的计数器初始化为n :new CountDownLatch(n),每当一个任务线程执行完毕,就将计数器减1 countdownlatch.countDown(),当计数器的值变为0时,在CountDownLatch上 await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时,主线程需要等待多个组件加载完毕,之后再继续执行。即可以通过初始化,定义线程个数。
  • 实现多个线程开始执行任务的最大并行性。注意是并行性,不是并发,强调的是多个线程在某一时刻同时开始执行。类似于赛跑,将多个线程放到起点,等待发令枪响,然后同时开跑。做法是初始化一个共享的 CountDownLatch 对象,将其计数器初始化为 1 :new CountDownLatch(1),多个线程在开始执行任务前首先 coundownlatch.await(),当主线程调用 countDown() 时,计数器变为0,多个线程同时被唤醒。

    源码分析

    /**
     * 构造器中的计数值(count)实际上就是闭锁需要等待的线程数量。这个值只能被设置一次,
     *而且CountDownLatch没有提供任何机制去重新设置这个计数值。
     *
     * @param count 在线程可以通过 @link #await 之前必须调用 @link #countDown 的次数
     * @throws IllegalArgumentException 如果给定参数小于0则抛出异常
     */
    public CountDownLatch(int count) 
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    
    /**
     * 使当前线程等待,直到计数器为零,除非当前线程被中断
     * 1.当前计数为零,则此方法立即返回
     * 2.如果当前计数大于零,则当前线程将因线程调度目的而被禁用并处于休眠状态,
     * 直到发生如下可能:
     *   由于调用了 @link #countDown 方法,计数达到零
     *   线程等待被中断。
     * @throws 如果当前线程在等待时被中断
     */
    public void await() throws InterruptedException 
        sync.acquireSharedInterruptibly(1);
    
    /**    
     * 可以设置等待时间,超过这个时间就会执行,不会等到计数器变为0,但是
     * 之前给定的线程还是会执行完
     *
     * @param timeout 等待时间长度
     * @param unit 等待时间单位
     * @return 
     * @throws 如果当前线程在等待时被中断
     */
    public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException 
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    

    与CountDownLatch的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。这样主线程的操作就会在这个方法上阻塞,直到其他线程完成各自的任务。
    其他N 个线程必须引用闭锁对象,因为他们需要通知CountDownLatch对象,他们已经完成了各自的任务。这种通知机制是通过 CountDownLatch.countDown()方法来完成的;每调用一次这个方法,在构造函数中初始化的count值就减1。所以当N个线程都调用了这个方法,count的值等于0,然后主线程就能通过await()方法,恢复执行自己的任务。
    注意:

  • CountDownLatch的构造函数 CountDownLatch countDownLatch = new CountDownLatch(7); //7表示需要等待执行完毕的线程数量。
  • 在每一个线程执行完毕之后,都需要执行 countDownLatch.countDown() 方法,不然计数器就不会准确;
  • 只有所有的线程执行完毕之后,才会执行 countDownLatch.await() 之后的 代码;
  • CountDownLatch 阻塞的是主线程;

    CountDownLatch 的使用示例

    @Slf4j
    public class CountDownLatchExample1 
    /**
     * 线程数量
     */
    private final static int threadCount = 200;
    
    public static void main(String[] args) throws Exception 
    
        ExecutorService exec = Executors.newCachedThreadPool();
    
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
    
        for (int i = 0; i < threadCount; i++) 
            final int threadNum = i;
            exec.execute(() -> 
                try 
                    test(threadNum);
                 catch (Exception e) 
                    log.error("exception", e);
                 finally 
                    // 表示一个请求已经完成
                    countDownLatch.countDown();
                
            );
        
        //使当前线程等待,直到计数器为零,除非当前线程被中断
        countDownLatch.await();
        //当这200个请求被处理完成之后,才会执行
        log.info("finish");
        exec.shutdown();
    
    
    private static void test(int threadNum) throws Exception 
        // 模拟请求的耗时操作
        Thread.sleep(100);
        log.info("", threadNum);
        Thread.sleep(100);
    
    

    上面的代码中,我们定义了请求的数量为200,当这200个请求被处理完成之后,才会执行System.out.println("finish");。

    //可以设置等待时间,超过这个时间就会执行,不会等到计数器变为0,但是之前给定的线程还是会执行完
    countDownLatch.await(20, TimeUnit.MILLISECONDS);

    上面代码中其他跟第一个代码一致,使用了wait设置等待一定时间后继续执行方法。

    
    /**
    * CountDownLatch 模拟并发调用多个任务
    *
    * @author zjq
    */
    @Slf4j
    public class CountDownLatchExample3 
    
    public static void main(String[] args) throws InterruptedException 
    
        CountDownLatch countDownLatch = new CountDownLatch(2) 
            @Override
            public void await() throws InterruptedException 
                super.await();
                log.info("其他线程执行完毕后主线程执行的内容");
                log.info("threadName:,", Thread.currentThread().getName() + " count down is ok");
            
        ;
    
        Thread thread1 = new Thread(new Runnable() 
            @Override
            public void run() 
                try 
                    Thread.sleep(1000);
                    log.info(Thread.currentThread().getName() + "任务已完成");
                 catch (InterruptedException e) 
                    e.printStackTrace();
                 finally 
                    //计数器减1
                    countDownLatch.countDown();
                
            
        , "thread111");
    
        Thread thread2 = new Thread(new Runnable() 
            @Override
            public void run() 
                try 
                    Thread.sleep(2000);
                    log.info(Thread.currentThread().getName() + "任务已完成");
                 catch (InterruptedException e) 
                    e.printStackTrace();
                 finally 
                    //计数器减1
                    countDownLatch.countDown();
                
            
        , "thread222");
    
        thread1.start();
        thread2.start();
    
        countDownLatch.await();
        log.info("====everything is end====");
    

上述代码执行输出结果:
```shell
[thread111] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - thread111任务已完成
[thread222] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - thread222任务已完成
[main] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - 其他线程执行完毕后主线程执行的内容
[main] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - threadName:main count down is ok,
[main] INFO com.zjq.concurrency.aqs.CountDownLatchExample3 - ====everything is end====
/**
 * CountDownLatch 模拟多个任务并发执行完毕后等待主线程发令同时执行后续操作
 *
 * @author zjq
 */
@Slf4j
public class CountDownLatchExample4 

    public static void main(String[] args) throws InterruptedException 

        CountDownLatch countDownLatch = new CountDownLatch(1);
        for (int i = 0; i < 5; i++) 
            new Thread(() -> 
                try 
                    //所有线程阻塞在这,等待主线程号令
                    log.info(Thread.currentThread().getName() + "已准备完毕!!");
                    countDownLatch.await();
                    log.info("【" + Thread.currentThread().getName() + "】" + "开始执行……");
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            ).start();
        
        // 主线程准备发令
        Thread.sleep(2000);
        log.info(Thread.currentThread().getName() + "发号施令,给我冲!!");
        // 主线程:执行发令
        countDownLatch.countDown();
    

实现最大并行性输出结果:

[Thread-3] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-3已准备完毕!!
[Thread-0] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-0已准备完毕!!
[Thread-2] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-2已准备完毕!!
[Thread-1] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-1已准备完毕!!
[Thread-4] INFO com.zjq.aqs.CountDownLatchExample4 - Thread-4已准备完毕!!
[main] INFO com.zjq.aqs.CountDownLatchExample4 - main发号施令,给我冲!!
[Thread-3] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-3】开始执行……
[Thread-1] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-1】开始执行……
[Thread-0] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-0】开始执行……
[Thread-2] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-2】开始执行……
[Thread-4] INFO com.zjq.aqs.CountDownLatchExample4 - 【Thread-4】开始执行……

CountDownLatch 的不足

CountDownLatch是一次性的,计数器的值只能在构造方法中初始化一次,之后没有任何机制再次对其设置值,当CountDownLatch使用完毕后,它不能再次被使用。

CountDownLatch应用场景

(1)实现最大的并行性:有时我们想同时启动多个线程,实现最大程度的并行性。例 如,我们想测试一个单例类。如果我们创建一个初始计数为1的CountDownLatch,并 让所有线程都在这个锁上等待,那么我们可以很轻松地完成测试。我们只需调用 一次 countDown()方法就可以让所有的等待线程同时恢复执行。
(2)开始执行前等待n个线程完成各自任务:例如应用程序启动类要确保在处理用户 请求前,所有N个外部系统已经启动和运行了。
(3)死锁检测:一个非常方便的使用场景是,你可以使用n个线程访问共享资源,在每次测试阶段的线程数目是不同的,并尝试产生死锁。

以上是关于AQS同步组件-CountDownLatch解析和案例的主要内容,如果未能解决你的问题,请参考以下文章

AQS 同步组件学习

5-2 AQS应用(组件)

AQS的几个同步组件

AQS解析与实战

Java 线程同步组件 CountDownLatch 与 CyclicBarrier 原理分析

[Java并发] AQS抽象队列同步器源码解析--锁获取过程