一、应用

与CountDownLatch类似，也是基于AQS，实现线程等待，达到指定线程数量后，线程被唤醒继续执行。
车站流水发车场景：
CyclicBarrier认为是一个车站，每辆车可乘坐人数为CyclicBarrier的阈值；把乘客认为是线程。
未达到阈值前，所有乘客在车中等待（线程阻塞）。
当线程达到阈值，则进行发车（即，唤醒线程继续执行）。然后阈值重置，重新等待到来线程达到阈值。

应用示例

  1.     //统计工厂员工平均工作量
  2.     static ThreadPoolExecutor executor = new ThreadPoolExecutor(
  3.             3,
  4.             5,
  5.             100,
  6.             TimeUnit.SECONDS,
  7.             new ArrayBlockingQueue<>(100),
  8.             (ThreadFactory) Thread::new);
  10.     private Map<String, Integer> workCount = new ConcurrentHashMap<>();
  12.     public static void main(String[] args) {
  14.         new CyclicBarrierDemo().count();
  16.     }
  19.     private CyclicBarrier cb = new CyclicBarrier(3, () -> {
  20.         int result = 0;
  21.         for (String key : workCount.keySet()) {
  22.             result += workCount.get(key);
  23.         }
  24.         System.out.println("工厂工人平均工作量为:" + (result / 3) );
  25.     });
  27.     private void count() {
  28.         for (int i = 0; i < 5; i++) {
  29.             executor.execute(() -> {
  30.                 //统计工人工作量
  31.                 int count = (int) (Math.random() * 40 + 60);
  32.                 workCount.put(Thread.currentThread().getName(), count);
  33.                 System.out.println(Thread.currentThread().getName()
  34.                         + "工人工作量：" + count);
  35.                 try {
  36.                     //执行完运行await(),等待所有学生平均成绩都计算完毕
  37.                     cb.await();
  38.                 } catch (InterruptedException | BrokenBarrierException e) {
  39.                     e.printStackTrace();
  40.                 }
  41.             });
  42.         }
  43.         executor.shutdown();
  44.     }
  45. }
  47. Thread-2工人工作量：92
  48. Thread-0工人工作量：89
  49. Thread-1工人工作量：95
  50. 工厂工人平均工作量为:92

二、原理分析

await()方法

  1. /**
  2. *
  3. * 如果当前线程不是最后一个到达的线程，则出于线程调度目的，将其休眠，
  4. *直到发生以下情况之一： 
  5. *     1.最后一个线程到达；（正常情况）
  6. *     2.其他线程中断当前线程；（当前线程抛出InterruptedException，并清除当前线程的中断状态；其他线程抛出BrokenBarrierException）
  7. *     3.另一个线程中断另一个等待的线程；
  8. *     4.其他线程在等待屏障时超时；（引发BrokenBarrierException）
  9. *     5.其他一些线程在此屏障上调用重置。
  10. **/    
  11.     public int await() throws InterruptedException, BrokenBarrierException {
  12.         try {
  13.             return dowait(false, 0L);
  14.         } catch (TimeoutException toe) {
  15.             throw new Error(toe); // cannot happen
  16.         }
  17.     }

dowait(false, 0L)核心逻辑

  2.     private int dowait(boolean timed, long nanos)
  3.         throws InterruptedException, BrokenBarrierException,
  4.                TimeoutException {
  5.         final ReentrantLock lock = this.lock;
  6.         lock.lock();
  7.         try {
  8.             final Generation g = generation;
  10.             //如果之前有线程被中断，则后续线程抛出BrokenBarrierException
  11.             if (g.broken)
  12.                 throw new BrokenBarrierException();
  14.             //检查当前线程是否被中断。中断则抛出InterruptedException。并使后续线程抛出BrokenBarrierException
  15.             if (Thread.interrupted()) {
  16.                 breakBarrier();//【1.】设置标识，使得后续线程抛出BrokenBarrierException；【2.】并唤醒所有已阻塞的线程
  17.                 throw new InterruptedException();
  18.             }
  20.             //count记录阈值，每到一个线程则【自减一】
  21.             int index = --count;
  23.             //如果是最后一个线程，(如果有)则执行初始化时，设置的命令
  24.             if (index == 0) {  // tripped
  25.                 boolean ranAction = false;
  26.                 try {
  27.                     final Runnable command = barrierCommand;
  28.                     if (command != null)
  29.                         command.run();
  30.                     ranAction = true;
  31.                     //唤醒所有阻塞线程。开始“新朝代”循环
  32.                     nextGeneration();
  33.                     return 0;
  34.                 } finally {
  35.                     //处理未知异常
  36.                     if (!ranAction)
  37.                         breakBarrier();
  38.                 }
  39.             }
  41.             // loop until tripped, broken, interrupted, or timed out
  42.             /**
  43.             * 执行到此处的线程
  44.             *   1.最后一个线程
  45.             *   2.非最后一个（会被阻塞）
  46.             **/
  47.             for (;;) {
  48.                 try {
  49.                     if (!timed)
  50.                         trip.await();//不是最后一个线程，则会被阻塞
  51.                     else if (nanos > 0L)
  52.                         nanos = trip.awaitNanos(nanos);
  53.                 } catch (InterruptedException ie) {
  54.                     if (g == generation && ! g.broken) {
  55.                         breakBarrier();
  56.                         throw ie;
  57.                     } else {
  58.                         // We're about to finish waiting even if we had not
  59.                         // been interrupted, so this interrupt is deemed to
  60.                         // "belong" to subsequent execution.
  61.                         Thread.currentThread().interrupt();
  62.                     }
  63.                 }
  65.                 if (g.broken)
  66.                     throw new BrokenBarrierException();
  68.                 //执行到此处线程。1.最后一个线程。2.重新被唤醒的线程
  69.                 //最后一个线程会开始“新朝代”循环，则g != generation为true，跳出循环
  70.                 if (g != generation)
  71.                     return index;
  73.                 if (timed && nanos <= 0L) {
  74.                     breakBarrier();
  75.                     throw new TimeoutException();
  76.                 }
  77.             }
  78.         } finally {
  79.             lock.unlock();
  80.         }
  81.     }

nextGeneration()方法

  1.     /**
  2.     *  唤醒所有阻塞线程。开始“新朝代”循环
  3.     *  private final Condition trip = lock.newCondition();
  4.     **/
  5.     private void nextGeneration() {
  6.         // signal completion of last generation
  7.         trip.signalAll();
  8.         // set up next generation
  9.         count = parties;
  10.         generation = new Generation();
  11.     }

await()方法

AbstractQueuedSynchronizer。加入条件队列被阻塞

  1.         /**
  2.         * 1.将当前节点添加到条件队列中。
  3.         * 2.并释放已经获取的锁。（state变量归零，并唤醒因争抢锁被阻塞的线程）。
  4.         * 3.调用park方法，阻塞当前线程
  5.         **/
  6.         public final void await() throws InterruptedException {
  7.             if (Thread.interrupted())
  8.                 throw new InterruptedException();
  10.             //添加的条件队列中（条件队列要素：firstWaiter、lastWaiter，条件队列中waitState=-2）
  11.             Node node = addConditionWaiter();
  12.             //释放已经持有的锁，并唤醒因争抢锁被阻塞的线程
  13.             int savedState = fullyRelease(node);
  14.             int interruptMode = 0;
  15.             //如果不在等待队列中，则进行阻塞
  16.             while (!isOnSyncQueue(node)) {
  17.                 LockSupport.park(this);
  18.                 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
  19.                     break;
  20.             }
  22.             //线程被唤醒后的逻辑。重新获取锁（独占锁逻辑）
  23.             if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
  24.                 interruptMode = REINTERRUPT;
  25.             if (node.nextWaiter != null) // clean up if cancelled
  26.                 unlinkCancelledWaiters();
  27.             if (interruptMode != 0)
  28.                 reportInterruptAfterWait(interruptMode);
  29.         }

signalAll()方法

AbstractQueuedSynchronizer。从条件队列中被唤醒

  1.         /**
  2.         *
  3.         * 1.将条件队列中元素，转移到等待队列中
  4.         * 2.并将条件队列中的waitstate。-2修改成0，最后修改成-1
  5.         **/
  6.         public final void signalAll() {
  8.             //如果当前线程，不是持有锁的线程，则抛出异常
  9.             if (!isHeldExclusively())
  10.                 throw new IllegalMonitorStateException();
  11.             Node first = firstWaiter;
  12.             if (first != null)
  13.                 doSignalAll(first);
  14.         }
  16.         //【循环】条件队列，转移到等待队列中
  17.          private void doSignalAll(Node first) {
  18.             lastWaiter = firstWaiter = null;
  19.             do {
  20.                 Node next = first.nextWaiter;
  21.                 first.nextWaiter = null;
  22.                 transferForSignal(first);
  23.                 first = next;
  24.             } while (first != null);
  25.         }
  27.     final boolean transferForSignal(Node node) {
  28.         /*
  29.          * 将-2修改为0
  30.          */
  31.         if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
  32.             return false;
  34.         /*
  35.          * 1.将节点插入等待队列尾部
  36.          * 2.将0修改为-1（等待当前线程释放锁，唤醒等待队列中元素）
  37.          *
  38.          * 异常情况下，才立即唤醒
  39.          */
  40.         Node p = enq(node);
  41.         int ws = p.waitStatus;
  42.         if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
  43.             LockSupport.unpark(node.thread);
  44.         return true;
  45.     }