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

Posted 2022-08-08 共饮一杯无

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个线程访问共享资源，在每次测试阶段的线程数目是不同的，并尝试产生死锁。