并发编程系列之CountDownLatch对战Cyclicbarrier

前言

前面我们介绍了并发容器和队列，今天我们来介绍几个非常有用的并发工具类，今天主要讲CountDownLatch和Cyclicbarrier这两个工具类，通过讲解并对比两个类的区别，OK，让我们开始今天的并发之旅吧。

什么是CountDownLatch？

CountDownLatch用于监听某些初始化操作，等待初始化执行完毕，通知主线程继续工作，允许一个或者多个线程等待其他线程完成操作。之前我们知道要实现线程等待还有一个方法就是jion方法，先让我们来回忆什么是Join方法：

Join用于让当前执行线程等待Join线程执行结束，实现原理是，不停的检查Join线程是否存活，如果存活则让当前线程永远等待下去，如果Join线程终止，则调用this.notifyAll方法唤醒等待的线程；

CountDownLatch其实也是来做这件事的，而且比Join更强大，使用起来也很轻便。

如何使用CountDownLatch？

我们看下面这个demo，看看如何使用CountDownLatch：

public static void main(String[] args) {
   // CountDownLatch接收一个int类型的计算器，此处是2代表计数器为2，意思是需要等待2个线程唤醒
   final CountDownLatch countDown = new CountDownLatch(2);
   
   Thread t1 = new Thread(new Runnable() {
     @Override
     public void run() {
       try {
         System.out.println("进入线程t1" + "等待其他线程处理完成...");
         // countDown.await()方法会阻塞当前线程即t1，没执行一次countDown()方法计数器就会-1
         // 直到计数器=0，则当前阻塞的线程t1被唤醒，继续执行
         countDown.await();
         System.out.println("t1线程继续执行...");
       } catch (InterruptedException e) {
         e.printStackTrace();
       }
     }
   },"t1");
   
   Thread t2 = new Thread(new Runnable() {
     @Override
     public void run() {
       try {
         System.out.println("t2线程进行初始化操作...");
         Thread.sleep(3000);
         System.out.println("t2线程初始化完毕，通知t1线程继续...");
         // 计数器-1
         countDown.countDown();
       } catch (InterruptedException e) {
         e.printStackTrace();
       }
     }
   });
   Thread t3 = new Thread(new Runnable() {
     @Override
     public void run() {
       try {
         System.out.println("t3线程进行初始化操作...");
         Thread.sleep(4000);
         System.out.println("t3线程初始化完毕，通知t1线程继续...");
         // 计数器再-1，唤醒t1，t1继续执行
         countDown.countDown();
       } catch (InterruptedException e) {
         e.printStackTrace();
       }
     }
   });
   t1.start();
   t2.start();
   t3.start();
 }

执行结果：

猜想：假设t1或者t2由于某某原因发生异常未能执行countDown.countDown()那么，t1线程岂不是要一直处于等待状态吗？当然JDK的设计大佬们才不会给你留下这么明显的问题呢，所以countDown还提供了一个

public boolean await(long timeout, TimeUnit unit)
       throws InterruptedException {
       return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
   }

，这个方法会在特定时间后，结束阻塞的线程。

CountDownLatch底层分析

我们主要看下CountDownLatch的await方法和countDown方法的源码，首先看看await源码：await内部采用公平锁来实现等待

public void await() throws InterruptedException {
       // 采用公平锁机制
       sync.acquireSharedInterruptibly(1);
}
public boolean await(long timeout, TimeUnit unit)
       throws InterruptedException {
       return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

再看下acquireSharedInterruptibly，这里只分析await，超时await原理也差不多：

public final void acquireSharedInterruptibly(int arg)
           throws InterruptedException {
       // 判断是否发生中断
       if (Thread.interrupted())
           throw new InterruptedException();
       // -1表示获取到了共享锁，1表示没有获取共享锁
       if (tryAcquireShared(arg) < 0)
           // 获取共享锁，继续执行
           doAcquireSharedInterruptibly(arg);
   }

private void doAcquireSharedInterruptibly(int arg)
       throws InterruptedException {
       final Node node = addWaiter(Node.SHARED);
       boolean failed = true;
       try {
           for (;;) {
               final Node p = node.predecessor();
               if (p == head) {
                   int r = tryAcquireShared(arg);
                   if (r >= 0) {
                       setHeadAndPropagate(node, r);
                       p.next = null; // help GC
                       failed = false;
                       return;
                   }
               }
               if (shouldParkAfterFailedAcquire(p, node) &&
                   parkAndCheckInterrupt())
                   throw new InterruptedException();
           }
       } finally {
           if (failed)
               cancelAcquire(node);
       }
   }

再看下countDown方法：

public void countDown() {
       // 每次释放一个计数器
       sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
       //尝试释放共享锁  
       if (tryReleaseShared(arg)) {
           doReleaseShared();
           return true;
       }
       return false;
   }


private void doReleaseShared() {
       
       for (;;) {
           Node h = head;
           if (h != null && h != tail) {
               int ws = h.waitStatus;
               if (ws == Node.SIGNAL) {
                   if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                       continue;            
              // 循环检查
                   unparkSuccessor(h);
               }
               else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                   continue;                
           // loop on failed CAS
           }
           if (h == head)                  
     // loop if head changed
               break;
       }
   }

什么是Cyclicbarrier？

Cyclicbarrier指的是可循环使用的屏障，主要是让一组线程到达一个屏障之后被阻塞，当最后一个线程到达时，屏障才会开门，所有被屏障拦截的线程才会继续干活。

如何使用Cyclicbarrier？

static class Runner implements Runnable {  
     private CyclicBarrier barrier;  
     private String name;  
     
     public Runner(CyclicBarrier barrier, String name) {  
         this.barrier = barrier;  
         this.name = name;  
     }  
     @Override  
     public void run() {  
         try {  
           // 因为是先打印后阻塞，所以这里getNumberWaiting的+1
           int numberWaiting = barrier.getNumberWaiting();
           int count = numberWaiting + 1 ;
           System.out.println(name + " 进入赛道，签到完毕，当前人数"+count);  
             barrier.await();  
         } catch (InterruptedException e) {  
             e.printStackTrace();  
         } catch (BrokenBarrierException e) {  
             e.printStackTrace();  
         }  
         System.out.println(name + " Go!!");  
     }  
 }
 
   public static void main(String[] args) throws IOException, InterruptedException {  
       CyclicBarrier barrier = new CyclicBarrier(10);
       // Executors是我们后续会讲的线程池
       ExecutorService executor = Executors.newFixedThreadPool(10);  
       
       for (int i = 100; i < 110; i++) {
         Thread.sleep(1000);  
          executor.submit(new Thread(new Runner(barrier, i+"号选手进场")));  
   }
       executor.shutdown();  
   }

执行结果：

某些情况下，我们需要让阻塞屏障解除的时候，某些线程需要先执行，例如某个运动员买通了裁判，比赛开始时，比别的选手提前开跑，当然这在现实比赛中是不允许的，此处我只是打个比方，对于这样的场景，Cyclicbarrier提供了：

public CyclicBarrier(int parties, Runnable barrierAction) {
       if (parties <= 0) throw new IllegalArgumentException();
       this.parties = parties;
       this.count = parties;
       this.barrierCommand = barrierAction;
   }

用于在线程到达屏障时，优先执行barrierAction线程；

Cyclicbarrier底层实现

public int await() throws InterruptedException, BrokenBarrierException {
       try {
           return dowait(false, 0L);
       } catch (TimeoutException toe) {
           throw new Error(toe); // cannot happen;
       }
   }


   private int dowait(boolean timed, long nanos)
       throws InterruptedException, BrokenBarrierException,
              TimeoutException {
       // 使用重入锁，同步进行wait操作，计数器+1
       final ReentrantLock lock = this.lock;
       lock.lock();
       try {
           final Generation g = generation;
           // 当前Generation处于打破状态,抛出异常
           if (g.broken)
               throw new BrokenBarrierException();
           // 当前Generation处于中断状态,抛出异常，并重置计数器，唤醒所有等待线程，可见见下面源码
           if (Thread.interrupted()) {
               breakBarrier();
               throw new InterruptedException();
           }

          int index = --count;
          // 当最后一个线程也到达了，就从调用中返回
          if (index == 0) {
              boolean ranAction = false;
              try {
                  final Runnable command = barrierCommand;
                  if (command != null)
                      command.run();
                  ranAction = true;
                  nextGeneration();
                  return 0;
              } finally {
                  // 如果运行command失败也会导致当前屏障被打破
                  if (!ranAction)
                      breakBarrier();
              }
          }

           // loop until tripped, broken, interrupted, or timed out
           for (;;) {
               try {
                   if (!timed)
                       trip.await();
                   else if (nanos > 0L)
                      // 挂起在条件变量的等待队列里，等待信号并自动释放锁
                       nanos = trip.awaitNanos(nanos);
               } catch (InterruptedException ie) {
                   // 如果当前线程被中断了则使得屏障被打破。并抛出异常
                   if (g == generation && ! g.broken) {
                       breakBarrier();
                       throw ie;
                   } else {
                       // We're about to finish waiting even if we had not
                       // been interrupted, so this interrupt is deemed to
                       // "belong" to subsequent execution.
                       Thread.currentThread().interrupt();
                   }
               }
               //从阻塞恢复之后，需要重新判断当前的状态
               if (g.broken)
                   throw new BrokenBarrierException();

               if (g != generation)
                   return index;

               if (timed && nanos <= 0L) {
                   breakBarrier();
                   throw new TimeoutException();
               }
           }
       } finally {
           lock.unlock();
       }
   }



private void breakBarrier() {
       generation.broken = true;
       count = parties;
       trip.signalAll();
   }

CountDownLatch和Cyclicbarrier比较

CountDownLatch就像一场跑步比赛，假设这场比赛有10个运动员，那么计数器初始值就为10，裁判员喊下比赛开始，就await阻塞在那，当每个运动员跑到终点就countDown一次，计数器-1，知道最后一个运动员到达终点即计数器为0，此时裁判员被唤醒，统计比赛结果，完成比赛。

Cyclicbarrier就像这场比赛时，裁判员首先准备好10条赛道，准备完毕就拿个小本子在那等着，每当以为选手到达赛道就签到一次，当10个选手全部签到完毕，裁判员就宣布比赛正式开始，继续执行下面的比赛。如果中间因为某某原因，某个选手未能到场或者天气原因，比赛推迟，签到信息就重置，比赛恢复之后选手需要重新签到；

区别总结：CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景，比如如果计算发生错误，可以重置计数器，并让线程们重新执行一次。