源码仓库Gitee · GitHub一.CountDownLatch1.概念public CountDownLatch(int count) {//初始化 if (count 0) throw new IllegalArgumentException(count 0); this.sync new Sync(count); }CountDownLatch是一个同步工具类用来协调多个线程之间的同步或者说起到线程之间的通信而不是用作互斥的作用。CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后计数器的值就会减一。当计数器的值为0时表示所有的线程都已经完成一些任务然后在CountDownLatch上等待的线程就可以恢复执行接下来的任务。下面有A、B、C、D4个线程同时执行A是主线程B、C、D是子线程A先开始执行后阻塞等待子线程全部执行结束才继续执行剩下的任务。2.用法1、某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为new CountDownLatch(n)每当一个任务线程执行完毕就将计数器减1 countdownLatch.countDown()当计数器的值变为0时在CountDownLatch上await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时主线程需要等待多个组件加载完毕之后再继续执行。public static void main(String[] args) { ExecutorService service Executors.newFixedThreadPool(3); final CountDownLatch latch new CountDownLatch(3); for (int i 0; i 3; i) { Runnable runnable new Runnable() { Override public void run() { try { System.out.println(子线程 Thread.currentThread().getName() 开始执行); Thread.sleep((long) (Math.random() * 10000)); System.out.println(子线程Thread.currentThread().getName()执行完成); latch.countDown();//当前线程调用此方法则计数减一 } catch (InterruptedException e) { e.printStackTrace(); } } }; service.execute(runnable); } try { System.out.println(主线程Thread.currentThread().getName()等待子线程执行完成...); latch.await();//阻塞当前线程直到计数器的值为0 System.out.println(主线程Thread.currentThread().getName()开始执行...); } catch (InterruptedException e) { e.printStackTrace(); } } 结果主线程main等待子线程执行完成...子线程pool-1-thread-1开始执行子线程pool-1-thread-3开始执行子线程pool-1-thread-2开始执行子线程pool-1-thread-3执行完成子线程pool-1-thread-1执行完成子线程pool-1-thread-2执行完成主线程main开始执行...2、实现多个线程开始执行任务的最大并行性。注意是并行性不是并发强调的是多个线程在某一时刻同时开始执行。类似于赛跑将多个线程放到起点等待发令枪响然后同时开跑。做法是初始化一个共享的CountDownLatch(1)将其计算器初始化为1多个线程在开始执行任务前首先countdownlatch.await()当主线程调用countDown()时计数器变为0多个线程同时被唤醒。public static void main(String[] args) { ExecutorService service Executors.newCachedThreadPool(); final CountDownLatch cdOrder new CountDownLatch(1); final CountDownLatch cdAnswer new CountDownLatch(4); for (int i 0; i 4; i) { Runnable runnable new Runnable() { Override public void run() { try { System.out.println(选手 Thread.currentThread().getName() 正在等待裁判发布口令); cdOrder.await(); System.out.println(选手 Thread.currentThread().getName() 已接受裁判口令); Thread.sleep((long) (Math.random() * 10000)); System.out.println(选手 Thread.currentThread().getName() 到达终点); cdAnswer.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } } }; service.execute(runnable); } try { Thread.sleep((long) (Math.random() * 10000)); System.out.println(裁判Thread.currentThread().getName()即将发布口令); cdOrder.countDown(); System.out.println(裁判Thread.currentThread().getName()已发送口令正在等待所有选手到达终点); cdAnswer.await(); System.out.println(所有选手都到达终点); System.out.println(裁判Thread.currentThread().getName()汇总成绩排名); } catch (InterruptedException e) { e.printStackTrace(); } service.shutdown(); } 结果:选手pool-1-thread-2正在等待裁判发布口令选手pool-1-thread-1正在等待裁判发布口令选手pool-1-thread-3正在等待裁判发布口令选手pool-1-thread-4正在等待裁判发布口令裁判main即将发布口令裁判main已发送口令正在等待所有选手到达终点选手pool-1-thread-2已接受裁判口令选手pool-1-thread-1已接受裁判口令选手pool-1-thread-3已接受裁判口令选手pool-1-thread-4已接受裁判口令选手pool-1-thread-2到达终点选手pool-1-thread-1到达终点选手pool-1-thread-4到达终点选手pool-1-thread-3到达终点所有选手都到达终点裁判main汇总成绩排名3.countDown解析递减锁存器的计数如果计数到达零则释放所有等待的线程。如果当前计数大于零则将计数减少.public void countDown() { sync.releaseShared(1); }countDown调用AQS的releaseSharedpublic final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//数量为0 doReleaseShared();//唤醒其他等待线程 return true; } return false; } protected boolean tryReleaseShared(int arg) {//releaseShared调用由CountDownLatch的内部类Sync实现 throw new UnsupportedOperationException(); } private static final class Sync extends AbstractQueuedSynchronizer {//CountDownLatch的内部类Sync protected boolean tryReleaseShared(int releases) { for (;;) {//自旋count不断-1直到为0则发起唤醒信号 int c getState();//获得数量在CountDownLatch(int count)初始化时定义了数量 if (c 0)//数量为0则返回false return false; int nextc c-1;//数量-1 //CAS更新状态nextc为0返回true if (compareAndSetState(c, nextc)) return nextc 0; } } }4.await解析使当前线程在锁存器倒计数至零之前一直等待除非线程被中断或超出了指定的等待时间。如果当前计数为零则唤醒阻塞线程。如果当前计数大于零则出于线程调度目的将禁用当前线程该线程将一直出于休眠状态public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1);//由AQS实现 } /** * Acquires in shared mode, aborting if interrupted. Implemented * by first checking interrupt status, then invoking at least once * {link #tryAcquireShared}, returning on success. Otherwise the * thread is queued, possibly repeatedly blocking and unblocking, * invoking {link #tryAcquireShared} until success or the thread * is interrupted. * 以共享模式获取如果中断被中止。 * 实现首先检查中断状态然后至少调用一次tryacquirered成功返回。 * 否则,线程排队可能会重复阻塞和取消阻塞 * 调用tryacquiremred直到成功或线程被打断了。 */ public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted())//有中断抛出异常 throw new InterruptedException(); if (tryAcquireShared(arg) 0)//CountDownLatch的Sync实现计数数量不为0表示有线程需要阻塞 doAcquireSharedInterruptibly(arg); } protected int tryAcquireShared(int acquires) { return (getState() 0) ? 1 : -1; } //以共享中断模式获取 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node addWaiter(Node.SHARED);//创建当前线程的节点并且锁的模型是共享锁将其添加到AQS CLH队列的末尾 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; } } //前继节点非head节点没资源获取将前继节点状态设置为SIGNAL通过park挂起node节点的线程 if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node);//结束该节点线程的请求 } }二.CyclicBarrier1.概念public CyclicBarrier(int parties, Runnable barrierAction) {//初始化 if (parties 0) throw new IllegalArgumentException(); this.parties parties; this.count parties; this.barrierCommand barrierAction; } public CyclicBarrier(int parties) {//初始化 this(parties, null); }允许一组线程全部等待彼此达到共同屏障点的同步辅助。 循环阻塞在涉及固定大小的线程方的程序中很有用这些线程必须偶尔等待彼此。 屏障被称为循环 因为它可以在等待的线程被释放之后重新使用。A CyclicBarrier支持一个可选的Runnable命令每个屏障点运行一次在派对中的最后一个线程到达之后但在任何线程释放之前。 在任何一方继续进行之前此屏障操作对更新共享状态很有用。CyclicBarrier当计数减少到0时会唤醒所有阻塞在同一个Condition上的线程与CountDownLatch不同的是所有的线程必须同时被唤醒就好比钓鱼比赛所有人必须同时开始抛竿一样。CountDownLatch只要求主线程的动作在其他依赖的线程执行完之后执行就OK。下面有A、B、C、D4个线程同时执行每个线程有任务a、b每个线程的每个任务执行完才开始继续下个任务执行。2.用法public class CyclicBarrierTest { public static void main(String[] args) { int count 10;//并发线程数 CyclicBarrier cyclicBarrier new CyclicBarrier(count); ExecutorService executorService Executors.newFixedThreadPool(count); int n 1; for (int i 0; i count; i) { executorService.execute(new Task(cyclicBarrier, n)); n; } executorService.shutdown(); // 关闭线程池 // 判断是否所有的线程已经运行完 while (!executorService.isTerminated()) { try { // 所有线程池中的线程执行完毕执行后续操作 System.out.println( is sleep); Thread.sleep(10000); System.out.println( is wake); } catch (InterruptedException e) { e.printStackTrace(); } } } } class Task implements Runnable { private CyclicBarrier cyclicBarrier; int n 0; public Task(CyclicBarrier cyclicBarrier, int n) { this.cyclicBarrier cyclicBarrier; this.n n; } Override public void run() { try { System.out.println(赛马 n 到达栅栏前); cyclicBarrier.await(); System.out.println(赛马 n 开始跑); cyclicBarrier.await(); System.out.println(赛马 n 到达终点); } catch (Exception e) { e.printStackTrace(); } } } 结果is sleep赛马2到达栅栏前赛马3到达栅栏前赛马4到达栅栏前赛马1到达栅栏前赛马5到达栅栏前赛马6到达栅栏前赛马7到达栅栏前赛马8到达栅栏前赛马9到达栅栏前赛马10到达栅栏前赛马10开始跑赛马3开始跑赛马2开始跑赛马4开始跑赛马1开始跑赛马6开始跑赛马7开始跑赛马5开始跑赛马9开始跑赛马8开始跑赛马8到达终点赛马2到达终点赛马3到达终点赛马4到达终点赛马7到达终点赛马10到达终点赛马5到达终点赛马6到达终点赛马1到达终点赛马9到达终点is wake3.await解析如果当前线程不是最后一个线程那么它被禁用以进行线程调度并且处于休眠状态直到发生下列事情之一最后一个线程到达; 要么一些其他线程当前线程为interrupts ; 要么一些其他线程interrupts其他等待线程之一; 要么一些其他线程在等待屏障时超时; 要么其他一些线程在这个屏障上调用reset()。CyclicBarrier的原理在CyclicBarrier的内部定义了一个Lock对象每当一个线程调用await方法时将拦截的线程数减1然后判断剩余拦截数是否为初始值parties如果不是进入Lock对象的条件队列等待。如果是执行barrierAction对象的Runnable方法然后将锁的条件队列中的所有线程放入锁等待队列中这些线程会依次的获取锁、释放锁。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 { final ReentrantLock lock this.lock;//获锁 lock.lock();//加锁 try { //当代每个屏障都会创建一个Generation实例 final Generation g generation; if (g.broken)//当代遭到破坏抛出异常 throw new BrokenBarrierException(); if (Thread.interrupted()) {//线程中断抛出异常 // 将损坏状态设置为true并通知其他阻塞在此栅栏上的线程 breakBarrier(); throw new InterruptedException(); } int index --count;//获取下标并-1 if (index 0) { //最后一个线程到达了 boolean ranAction false; try { final Runnable command barrierCommand; if (command ! null) command.run();//执行栅栏任务 ranAction true; nextGeneration();// 更新一代将count重置将generation重置唤醒之前等待的线程 return 0; } finally { // 如果执行栅栏任务的时候失败了就将损坏状态设置为true 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 {// 上面条件不满足说明这个线程不是这代的就不会影响当前这代栅栏的执行所以就打个中断标记 Thread.currentThread().interrupt(); } } // 当有任何一个线程中断了就会调用breakBarrier方法 //就会唤醒其他的线程其他线程醒来后也要抛出异常 if (g.broken) throw new BrokenBarrierException(); // g ! generation表示正常换代了返回当前线程所在栅栏的下标 // 如果 g generation说明还没有换代那为什么会醒了 // 因为一个线程可以使用多个栅栏当别的栅栏唤醒了这个线程就会走到这里所以需要判断是否是当前代。 // 正是因为这个原因才需要generation来保证正确。 if (g ! generation) return index; // 如果有时间限制且时间小于等于0销毁栅栏并抛出异常 if (timed nanos 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }dowait(boolean, long)方法的主要逻辑处理比较简单如果该线程不是最后一个调用await方法的线程则它会一直处于等待状态除非发生以下情况最后一个线程到达即index 0某个参与线程等待超时某个参与线程被中断调用了CyclicBarrier的reset()方法。该方法会将屏障重置为初始状态在上面的源代码中我们可能需要注意Generation 对象在上述代码中我们总是可以看到抛出BrokenBarrierException异常那么什么时候抛出异常呢如果一个线程处于等待状态时如果其他线程调用reset()或者调用的barrier原本就是被损坏的则抛出BrokenBarrierException异常。同时任何线程在等待时被中断了则其他所有线程都将抛出BrokenBarrierException异常并将barrier置于损坏状态。同时Generation描述着CyclicBarrier的更新换代。在CyclicBarrier中同一批线程属于同一代。当有parties个线程到达barrier之后generation就会被更新换代。其中broken标识该当前CyclicBarrier是否已经处于中断状态。注意事项CyclicBarrier使用独占锁来执行await方法并发性可能不是很高。如果在等待过程中线程被中断了就抛出异常。但如果中断的线程所对应的CyclicBarrier不是这代的比如在最后一次线程执行signalAll后并且更新了这个“代”对象。在这个区间这个线程被中断了那么JDK认为任务已经完成了就不必在乎中断了只需要打个标记。该部分源码已在dowait(boolean, long)方法中进行了注释。如果线程被其他的CyclicBarrier唤醒了那么g肯定等于generation这个事件就不能return了而是继续循环阻塞。反之如果是当前CyclicBarrier唤醒的就返回线程在CyclicBarrier的下标。完成了一次冲过栅栏的过程。该部分源码已在dowait(boolean, long)方法中进行了注释。参考CyclicBarrierhttps://blog.csdn.net/qq_38293564/article/details/80558157