Java并发包学习(CountDownLatch,Seamphore,CyclicBarrier,Exchanger)

我们常说的“并发包”指的是java.util.concurrent这个包，后面简称 J.U.C，里面包含大量多线程和并发编程的工具。J.U.C 包是 JDK 1.5 版本引入，本文所有内容基于 JDK11。

工具类

工具包内主要有这几个工具：计数器CountDownLatch、回环栅栏（光看名字估计一头雾水）CyclicBarrier、信号量Semaphore、创建线程池的Executors，最后一个交换数据的Exchanger。本章几个工具的原理都是基于 AQS，关于 AQS 和ReentrantLock，都有单独的文章解读，就不专门介绍了。

CountDownLatch

一个计数器，常见这样一种场景：多个任务分发个不同的线程去执行，全部执行完毕后回到主线程。当然有不同的实现方式，用CountDownLatch实现起来就很简单：

int taskNum = 20;CountDownLatch countLatch = new CountDownLatch(taskNum);for (int i = 0; i < taskNum; i++) {new Thread(() -> {try {Thread.sleep(200); //假装做了点什么System.out.println(Thread.currentThread().getName() + "执行完毕");} catch (InterruptedException e) {e.printStackTrace();}countLatch.countDown();  // 减1操作}).start();}try {countLatch.await(); // 阻塞线程，直至countDown方法调用次数等于taskNum，才会释放线程。for循环次数必须大于等于taskNumSystem.out.println("全部任务执行完毕");} catch (InterruptedException e) {e.printStackTrace();}

Seamphore

信号量，类似于控制并发的时候用到的“令牌桶”算法，通过控制信号总数，不断释放和回收信号来控制并发数量。假设有这么一个场景，假设我们有五个通道可以执行任务，任务总数是 40，所以同一时刻只能有最多五个线程执行，其余的要等待，因此我们使用Seamphore来不断方法许可和收回许可：

Semaphore semaphore = new Semaphore(5);for (int i = 0; i < 40; i++) {new Thread(() -> {try {semaphore.acquire();  // 获取许可，获取失败则进入阻塞System.out.println(Thread.currentThread().getName() + "取得许可，开始执行任务");Thread.sleep(new Random().nextInt(2000));System.out.println(Thread.currentThread().getName() + "任务完成，释放许可");semaphore.release();  // 释放许可。释放后激活阻塞线程去争取许可} catch (InterruptedException e) {e.printStackTrace();}}).start();}

CyclicBarrier

回环栅栏光看名字不好理解，其实作用很上面的计数器类似，有点类似于 Java 虚拟机中的“安全点”。执行一个任务的时候，所有线程到达栅栏之后停止，等待所有其他线程都到达这个点，然后一起进入下一阶段。与计数器不同的是，CyclicBarrier可以重复使用，举个栗子：
用于协调多个线程同步执行操作的场合，所有线程等待完成,然后一起做事情( 相互之间都准备好,然后一起做事情 )
例如百米赛跑，必须等待所有运动员都准备好了，才能比赛。

CyclicBarrier cyclicBarrier = new CyclicBarrier(20);for (int i = 0; i < 20; i++) {new Thread(() -> {long timeStamp = System.currentTimeMillis();try {Thread.sleep(new Random().nextInt(2000));System.out.println(Thread.currentThread().getName() + ":一阶段任务完成，花费了" + (System.currentTimeMillis() - timeStamp) + "毫秒，开始等待其他线程");cyclicBarrier.await();System.out.println(Thread.currentThread().getName() + ":所有线程执行完成，开始下一阶段");timeStamp = System.currentTimeMillis();Thread.sleep(new Random().nextInt(2000));System.out.println(Thread.currentThread().getName() + ":二阶段任务完成，花费了" + (System.currentTimeMillis() - timeStamp) + "毫秒，开始等待其他线程");cyclicBarrier.await();System.out.println(Thread.currentThread().getName() + ":所有线程任务完成");} catch (InterruptedException | BrokenBarrierException e) {e.printStackTrace();}}).start();}

Exchanger

交换器顾名思义，就是用来交换数据，理解和使用起来是最简单的，但是内部实现很精巧复杂。使用很简单，只有两个方法，作用就是两个线程在一个安全点交换数据，产生数据慢的那个会阻塞等待。

Exchanger<Integer> exchanger = new Exchanger<>();new Thread(() -> {int num = new Random().nextInt(1000);System.out.println("交换之前：Thread1:" + num);try {num = exchanger.exchange(num);System.out.println("交换完毕：Thread1:" + num);} catch (InterruptedException e) {e.printStackTrace();}}).start();new Thread(() -> {int num = new Random().nextInt(1000);System.out.println("交换之前：Thread2:" + num);try {Thread.sleep(2000);num = exchanger.exchange(num);System.out.println("交换完毕：Thread2:" + num);} catch (InterruptedException e) {e.printStackTrace();}}).start();