本篇文章主要记录了 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 ();
}
}
|
