本篇文章主要记录了 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 方法核心实现如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31
| public void countDown() { sync.releaseShared (1); }
public final boolean releaseShared(int arg) { if (tryReleaseShared (arg)) { doReleaseShared (); return true; } return false; }
private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus (h, Node.SIGNAL, 0)) continue; unparkSuccessor (h); } else if (ws == 0 && !compareAndSetWaitStatus (h, 0, Node.PROPAGATE)) continue; } if (h == head) break; } }
|
CountDownlatch 的使用案例,下面使用三个线程来打印三个 List,三个线程任务都完成得时候才输出 Print Task Finish!
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37
| 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 (); }).start (); new Thread(()->{ for (String string: list2) { System.out.println (Thread.currentThread ().getName () + ":" + string); } countDownLatch.countDown (); }).start (); new Thread(()->{ for (String string: list3) { System.out.println (Thread.currentThread ().getName () + ":" + string); } countDownLatch.countDown (); }).start (); try { countDownLatch.await (); } catch (InterruptedException e) { e.printStackTrace (); } System.out.println ("Print Task Finish!"); } }
|

栅栏 CyclicBarrier
CyclicBarrier:阻塞当前线程,等待其他线程。等待其它线程,且会阻塞自己当前线程,所有线程必须同时到达栅栏位置后才能继续执行;所有线程到达栅栏处,可以触发执行另外一个预先设置的线程。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43
| 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 { 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 (" 线程 & quot; + Thread.currentThread ().getName () + " 已经送达 & quot; + System.currentTimeMillis ()); try { cyclicBarrier.await (); } catch (BrokenBarrierException e) { e.printStackTrace (); } catch (InterruptedException e) { e.printStackTrace (); } System.out.println (" 线程 & quot; + Thread.currentThread ().getName () + " 开始处理 & quot; + System.currentTimeMillis ()); } } }
|

信号量 Semaphore
Semaphore 也叫信号量,在 JDK1.5 被引入, 可以用来控制同时访问特定资源的线程数量 ,通过协调各个线程,以保证合理的使用资源。Semaphore 内部维护了一组虚拟的许可,许可的数量可以通过构造函数的参数指定。
访问特定资源前,必须使用 acquire 方法获得许可,如果许可数量为 0,该线程则一直阻塞,直到有可用许可。访问资源后,使用 release 释放许可。Semaphore 和 ReentrantLock 类似,获取许可有公平策略和非公平许可策略,默认情况下使用非公平策略。
信号量 Semaphore 得应用场景:Semaphore 可以用来做流量分流,特别是对公共资源有限的场景,比如数据库连接。假设有这个的需求,读取几万个文件的数据到数据库中,由于文件读取是 IO 密集型任务,可以启动几十个线程并发读取,但是数据库连接数只有 10 个,这时就必须控制最多只有 10 个线程能够拿到数据库连接进行操作。这个时候,就可以使用 Semaphore 做流量控制。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31
| package thread_study;
import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore;
public class SemaphoreDemo { public static void main(String [] args) { ExecutorService pool = Executors.newCachedThreadPool (); Semaphore semaphore = new Semaphore(5); for (int i = 0; i < 20; i++) { final int NO = i; pool.execute (()->{ try { semaphore.acquire (); System.out.println ("Accessing: " + NO); Thread.sleep ((long)(Math.random () * 10000)); semaphore.release (); } catch (InterruptedException e) { e.printStackTrace (); } }); } pool.shutdown (); } }
|

交换器 Exchanger
Exchanger(交换者)是一个用于线程间数据交换协作的工具类。它提供一个同步点,在这个同步点多个线程间两两之间线程可以交换彼此的数据。这两个线程通过 exchange 方法交换数据, 如果第一个线程先执行 exchange 方法,它会一直等待第二个线程也执行 exchange 方法,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
| import java.util.concurrent.Exchanger; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit;
public class ExchangerDemo { public static void main(String [] args) { Exchanger<String> exchanger = new Exchanger<>(); ExecutorService threadPool = Executors.newFixedThreadPool (2); threadPool.execute (()->{ try { String girl = exchanger.exchange (" 我其实暗恋你很久了...."); System.out.println (" 女生说: " + girl); } catch (InterruptedException e) { e.printStackTrace (); } });
threadPool.execute (()->{ try { System.out.println (" 女生慢慢的从教室走出来.... "); TimeUnit.SECONDS.sleep (3); String boy = exchanger.exchange (" 我很喜欢你...."); System.out.println (" 男生说:" + boy); } catch (InterruptedException e) { e.printStackTrace (); } });
threadPool.shutdown (); } }
|
