JUC的四个并发工具类
本篇文章主要记录了JUC的四个并发工具类，闭锁CountDownlatch、栅栏CyclicBarrier、信号量Semaphore、交换器Exchanger。CountDownlatch通常用于主线程等待其他任务线程执行完毕的场景；CyclicBarrier主要阻塞当前线程，等待其他线程（大家无论谁先跑到A点，必须要等其他线程也到达了A点，大家才能继续）。信号量Semaphore可以用来控制同时访问特定资源的线程数量（比如100个线程只能有10个线程可以获得MySQL连接）。交换器Exchanger很少用，只适用于两个线程在同步点交换数据的场景（如下图）。
闭锁CountDownlatch
CountDownLatch也叫闭锁，使得一(多)个主线程必须等待其他线程完成操作后再执行。 CountDownLatch内部维护一个计数器(父类的int state)，主线程先执行await方法，如果此时计数器大于0，则阻塞等待。当一个线程完成任务后，计数器值减1。直到计数器为0时，表示所有的线程已经完成任务，等待的主线程被唤醒继续执行。
CountDownLatch实现主要基于Java同步器AQS，关于AQS的分析可以看我这篇文章： 《ReentrantLock与AQS》 ，countDown方法核心实现如下：
1public void countDown() {
2 sync.releaseShared(1);
3}
4
5public final boolean releaseShared(int arg) {
6 // countDownLatch -> tryReleaseShared
7 if (tryReleaseShared(arg)) {
8 doReleaseShared();
9 return true;
10 }
11 return false;
12}
13
14private void doReleaseShared() {
15 for (;;) {
16 Node h = head;
17 if (h != null && h != tail) {
18 int ws = h.waitStatus;
19 if (ws == Node.SIGNAL) {
20 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
21 continue; // loop to recheck cases
22 unparkSuccessor(h);
23 }
24 else if (ws == 0 &&
25 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
26 continue; // loop on failed CAS
27 }
28 if (h == head) // loop if head changed
29 break;
30 }
31}
CountDownlatch的使用案例，下面使用三个线程来打印三个List，三个线程任务都完成得时候才输出Print Task Finish！
1import java.util.Arrays;
2import java.util.List;
3import java.util.concurrent.CountDownLatch;
4
5public class CountDownLatchDemo {
6 public static void main(String[] args) {
7 List<String> list1 = Arrays.asList("AAA", "BBB", "CCC");
8 List<String> list2 = Arrays.asList("DDD", "EEE", "FFF");
9 List<String> list3 = Arrays.asList("GGG", "HHH", "III");
10
11 CountDownLatch countDownLatch = new CountDownLatch(3);
12 new Thread(()->{
13 for (String string: list1) {
14 System.out.println(Thread.currentThread().getName() + ":" + string);
15 }
16 countDownLatch.countDown();
17 }).start();
18 new Thread(()->{
19 for (String string: list2) {
20 System.out.println(Thread.currentThread().getName() + ":" + string);
21 }
22 countDownLatch.countDown();
23 }).start();
24 new Thread(()->{
25 for (String string: list3) {
26 System.out.println(Thread.currentThread().getName() + ":" + string);
27 }
28 countDownLatch.countDown();
29 }).start();
30 try {
31 countDownLatch.await();
32 } catch (InterruptedException e) {
33 e.printStackTrace();
34 }
35 System.out.println("Print Task Finish！");
36 }
37}
栅栏CyclicBarrier
CyclicBarrier：阻塞当前线程，等待其他线程。等待其它线程，且会阻塞自己当前线程，所有线程必须同时到达栅栏位置后才能继续执行；所有线程到达栅栏处，可以触发执行另外一个预先设置的线程。
1import java.util.concurrent.BrokenBarrierException;
2import java.util.concurrent.CyclicBarrier;
3
4public class CyclicBarrierDemo {
5 public static void main(String[] args) throws InterruptedException {
6 new CyclicBarrierDemo().go();
7 }
8
9 private void go() throws InterruptedException {
10 //初始化栅栏得参与者数为3
11 CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
12 new Thread(new Task(cyclicBarrier), "Thread1").start();
13 Thread.sleep(1000);
14 new Thread(new Task(cyclicBarrier), "Thread2").start();
15 Thread.sleep(1000);
16 new Thread(new Task(cyclicBarrier), "Thread3").start();
17 }
18
19 class Task implements Runnable{
20 private CyclicBarrier cyclicBarrier;
21
22 public Task(CyclicBarrier cyclicBarrier) {
23 this.cyclicBarrier = cyclicBarrier;
24 }
25
26 @Override
27 public void run() {
28 System.out.println("线程" + Thread.currentThread().getName()
29 + "已经送达"
30 + System.currentTimeMillis());
31 try {
32 cyclicBarrier.await();
33 } catch (BrokenBarrierException e) {
34 e.printStackTrace();
35 } catch (InterruptedException e) {
36 e.printStackTrace();
37 }
38 System.out.println("线程" + Thread.currentThread().getName()
39 + "开始处理"
40 + System.currentTimeMillis());
41 }
42 }
43}
信号量Semaphore
Semaphore也叫信号量，在JDK1.5被引入，可以用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用资源。Semaphore内部维护了一组虚拟的许可，许可的数量可以通过构造函数的参数指定。
访问特定资源前，必须使用acquire方法获得许可，如果许可数量为0，该线程则一直阻塞，直到有可用许可。访问资源后，使用release释放许可。Semaphore和ReentrantLock类似，获取许可有公平策略和非公平许可策略，默认情况下使用非公平策略。
信号量Semaphore得应用场景：Semaphore可以用来做流量分流，特别是对公共资源有限的场景，比如数据库连接。假设有这个的需求，读取几万个文件的数据到数据库中，由于文件读取是IO密集型任务，可以启动几十个线程并发读取，但是数据库连接数只有10个，这时就必须控制最多只有10个线程能够拿到数据库连接进行操作。这个时候，就可以使用Semaphore做流量控制。
1package thread_study;
2
3import java.util.concurrent.ExecutorService;
4import java.util.concurrent.Executors;
5import java.util.concurrent.Semaphore;
6
7public class SemaphoreDemo {
8 public static void main(String[] args) {
9 ExecutorService pool = Executors.newCachedThreadPool();
10 //只能5个线程同时访问
11 Semaphore semaphore = new Semaphore(5);
12 //模拟20个客户端访问
13 for (int i = 0; i < 20; i++) {
14 final int NO = i;
15 pool.execute(()->{
16 try {
17 //获取许可
18 semaphore.acquire();
19 System.out.println("Accessing: " + NO);
20 Thread.sleep((long)(Math.random() * 10000));
21 //访问完毕后释放
22 semaphore.release();
23 } catch (InterruptedException e) {
24 e.printStackTrace();
25 }
26 });
27 }
28 //退出线程池
29 pool.shutdown();
30 }
31}
交换器Exchanger
Exchanger（交换者）是一个用于线程间数据交换协作的工具类。它提供一个同步点，在这个同步点多个线程间两两之间线程可以交换彼此的数据。这两个线程通过exchange方法交换数据， 如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。
1import java.util.concurrent.Exchanger;
2import java.util.concurrent.ExecutorService;
3import java.util.concurrent.Executors;
4import java.util.concurrent.TimeUnit;
5
6public class ExchangerDemo {
7 public static void main(String[] args) {
8 Exchanger<String> exchanger = new Exchanger<>();
9 ExecutorService threadPool = Executors.newFixedThreadPool(2);
10 threadPool.execute(()->{
11 try {
12 //男生对女生说的话
13 String girl = exchanger.exchange("我其实暗恋你很久了....");
14 System.out.println("女生说: " + girl);
15 } catch (InterruptedException e) {
16 e.printStackTrace();
17 }
18 });
19
20 threadPool.execute(()->{
21 try {
22 System.out.println("女生慢慢的从教室走出来.... ");
23 TimeUnit.SECONDS.sleep(3);
24 //男生对女生说的话
25 String boy = exchanger.exchange("我很喜欢你....");
26 System.out.println("男生说：" + boy);
27 } catch (InterruptedException e) {
28 e.printStackTrace();
29 }
30 });
31
32 threadPool.shutdown();
33 }
34}
