JUC并发包下有四个并发工具类，闭锁CountDownlatch、栅栏CyclicBarrier、信号量Semaphore、交换器Exchanger。
- CountDownlatch通常用于主线程等待其他任务线程执行完毕的场景，类似于Join
- CyclicBarrier主要阻塞当前线程，等待其他线程(大家无论谁先跑到A点，必须要等其他线程也到达了A点，大家才能继续)。
- 信号量Semaphore可以用来控制同时访问特定资源的线程数量(比如100个线程只能有10个线程可以获得MySQL连接)。
- 交换器Exchanger很少用，只适用于两个线程在同步点交换数据的场景
接下来分别详细的介绍下这四种工具类。
CountDownlatch类
CountDownLatch是一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待，CountDownLatch也叫闭锁，使得一(多)个主线程必须等待其他线程完成操作后再执行。
private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; Sync(int count) { setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } }
CountDownLatch内部维护一个Sync 类，包含计数器(继承自AQS的state)，主线程先执行await方法，如果此时计数器大于0，则阻塞等待。当一个线程完成任务后，计数器值减1。直到计数器为0时，表示所有的线程已经完成任务，等待的主线程被唤醒继续执行
import java.util.Arrays; import java.util.List; import java.util.concurrent.CountDownLatch; public class CountDownLatchDemo { public static void main(String[] args) { List<String> list1 = Arrays.asList("AAA", "BBB", "CCC"); List<String> list2 = Arrays.asList("DDD", "EEE", "FFF"); List<String> list3 = Arrays.asList("GGG", "HHH", "III"); CountDownLatch countDownLatch = new CountDownLatch(3); new Thread(()->{ for (String string: list1) { System.out.println(Thread.currentThread().getName() + ":" + string); } countDownLatch.countDown(); //锁减去1 }).start(); new Thread(()->{ for (String string: list2) { System.out.println(Thread.currentThread().getName() + ":" + string); } countDownLatch.countDown(); //锁减去1 }).start(); new Thread(()->{ for (String string: list3) { System.out.println(Thread.currentThread().getName() + ":" + string); } countDownLatch.countDown(); //锁减去1，为0 }).start(); try { countDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Print Task Finish！"); } }
返回结果为：
Thread-0:AAA Thread-0:BBB Thread-0:CCC Thread-1:DDD Thread-1:EEE Thread-1:FFF Thread-2:GGG Thread-2:HHH Thread-2:III Print Task Finish！
使用三个线程来打印三个List，三个线程任务都完成得时候才允许主线程继续允许输出Print Task Finish！
CyclicBarrier类
阻塞当前线程，等待其他线程。等待其它线程，且会阻塞自己当前线程，所有线程必须同时到达栅栏位置后才能继续执行；所有线程到达栅栏处，可以触发执行另外一个预先设置的线程
import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CyclicBarrier; public class CyclicBarrierDemo { public static void main(String[] args) throws InterruptedException { new CyclicBarrierDemo().go(); } private void go() throws InterruptedException { //初始化栅栏得参与者数为3 CyclicBarrier cyclicBarrier = new CyclicBarrier(3); new Thread(new Task(cyclicBarrier), "Thread1").start(); Thread.sleep(1000); new Thread(new Task(cyclicBarrier), "Thread2").start(); Thread.sleep(1000); new Thread(new Task(cyclicBarrier), "Thread3").start(); } class Task implements Runnable{ private CyclicBarrier cyclicBarrier; public Task(CyclicBarrier cyclicBarrier) { this.cyclicBarrier = cyclicBarrier; } @Override public void run() { System.out.println("线程" + Thread.currentThread().getName() + "已经送达" + System.currentTimeMillis()); try { cyclicBarrier.await(); //栅栏唤醒，拦住执行线程 } catch (BrokenBarrierException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("线程" + Thread.currentThread().getName() + "开始处理" + System.currentTimeMillis()); } } }
返回结果为：
线程Thread1已经送达1614786037856 线程Thread2已经送达1614786038856 线程Thread3已经送达1614786039857 线程Thread3开始处理1614786039857 线程Thread1开始处理1614786039857 线程Thread2开始处理1614786039857
可以看到只有三个线程都到达了，才开始处理，CyclicBarrier和CountDownLatch的区别是：
- CountDownLatch的作用是允许1或N个线程等待其他线程完成执行；而CyclicBarrier则是允许N个线程相互等待
- CountDownLatch的计数器无法被重置；CyclicBarrier的计数器可以被重置后使用，因此它被称为是循环的barrier
可以看出二者场景不同。
Semaphore类
Semaphore也叫信号量，可以用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用资源。Semaphore内部维护了一组虚拟的许可，许可的数量可以通过构造函数的参数指定。
- 访问特定资源前，必须使用acquire方法获得许可，如果许可数量为0，该线程则一直阻塞，直到有可用许可。
- 访问资源后，使用release释放许可。
- Semaphore和ReentrantLock类似，获取许可有公平策略和非公平许可策略，默认情况下使用非公平策略。
信号量Semaphore得应用场景：Semaphore可以用来做流量分流，特别是对公共资源有限的场景，比如数据库连接。假设有这个的需求，读取几万个文件的数据到数据库中，由于文件读取是IO密集型任务，可以启动几十个线程并发读取，但是数据库连接数只有10个，这时就必须控制最多只有10个线程能够拿到数据库连接进行操作。这个时候，就可以使用Semaphore做流量控制
package com.company; import java.util.concurrent.*; public class ThreadTest { public static void main(String[] args) throws InterruptedException { ExecutorService pool = Executors.newCachedThreadPool(); //只能5个线程同时访问 Semaphore semaphore = new Semaphore(5); //模拟20个客户端访问 for (int i = 0; i < 20; i++) { final int NO = i; pool.execute(()->{ try { //获取许可 semaphore.acquire(); System.out.println("Accessing: " + NO); Thread.sleep(5000); //访问完毕后释放 System.out.println("每隔5秒释放出5个信号量供线程使用" ); Thread.sleep(5000); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } }); } //退出线程池 pool.shutdown(); } }
返回结果如下：
Accessing: 1 Accessing: 4 Accessing: 0 Accessing: 2 Accessing: 3 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 Accessing: 5 Accessing: 8 Accessing: 7 Accessing: 6 Accessing: 9 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 Accessing: 10 Accessing: 11 Accessing: 12 Accessing: 13 Accessing: 14 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 Accessing: 16 Accessing: 17 Accessing: 15 Accessing: 18 Accessing: 19 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用 每隔5秒释放出5个信号量供线程使用
semaphore.acquire();表示开启信号量限制，在限制被解除前，一次只能有5个线程能活动，即使开启了20个线程。信号量维护了一个信号量许可集。线程可以通过调用acquire()来获取信号量的许可；当信号量中有可用的许可时，线程能获取该许可；否则线程必须等待，直到有可用的许可为止。 线程可以通过release()来释放它所持有的信号量许可
Exchanger类
Exchanger(交换者)是一个用于线程间数据交换协作的工具类。它提供一个同步点，在这个同步点多个线程间两两之间线程可以交换彼此的数据。这两个线程通过exchange方法交换数据， 如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方
public static void main(String[] args) { public static void main(String[] args) { Exchanger<String> exchanger = new Exchanger<>(); ExecutorService threadPool = Executors.newFixedThreadPool(2); threadPool.execute(()->{ try { //女生对男生说的话 TimeUnit.SECONDS.sleep(5); System.out.println("Girl快一点"); String girl = exchanger.exchange("hi girl"); System.out.println("girl said: " + girl); } catch (InterruptedException e) { e.printStackTrace(); } }); threadPool.execute(()->{ try { TimeUnit.SECONDS.sleep(3); System.out.println("BOY慢一点"); //男生对女生说的话 String boy = exchanger.exchange("hi boy"); System.out.println("boy said：" + boy); } catch (InterruptedException e) { e.printStackTrace(); } }); threadPool.shutdown(); }
返回结果为：
BOY慢一点 Girl快一点 //获得了交换打印值 girl said: hi boy boy said：hi girl
可以看的出在exchange相交之前，BOY慢一点打印完之后没有执行hi girl，而是等待Girl快一点所属线程也执行到了exchanger 点之后两个线程才一起打印。
