一、简介
CyclicBarrier 是一个同步工具类，它允许一组线程在到达某个栅栏点(common barrier point)互相等待，发生阻塞，直到最后一个线程到达栅栏点，栅栏才会打开，处于阻塞状态的线程恢复继续执行.它非常适用于一组线程之间必需经常互相等待的情况。CyclicBarrier 字面理解是循环的栅栏，之所以称之为循环的是因为在等待线程释放后，该栅栏还可以复用。
建议阅读 CyclicBarrier 源码前，先深入研究一下 ReentrantLock 的原理，搞清楚 Condition 里 await 和 signal 的原理。
二、应用场景
使用 CyclicBarrier 来模拟一下对战平台中玩家需要完全准备好了才能进入游戏的场景，例如 LOL 的加载过程。
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
| public class CyclicBarrierTest {
private final static ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(5);
private final static CyclicBarrier BARRIER = new CyclicBarrier(5);
public static void main(String[] args) {
for (int i = 1; i <= 5; i++) {
final String name = "玩家" + i;
EXECUTOR_SERVICE.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(2000);
System.out.println(name + "已准备,等待其他玩家准备...");
BARRIER.await();
Thread.sleep(1000);
System.out.println(name + "已加入游戏");
} catch (InterruptedException | BrokenBarrierException e) {
System.out.println(name + "离开游戏");
}
}
});
}
EXECUTOR_SERVICE.shutdown();
}
}
|
输出结果：
1
2
3
4
5
6
7
8
9
10
| 玩家2已准备,等待其他玩家准备...
玩家4已准备,等待其他玩家准备...
玩家3已准备,等待其他玩家准备...
玩家1已准备,等待其他玩家准备...
玩家5已准备,等待其他玩家准备...
玩家4已加入游戏
玩家1已加入游戏
玩家2已加入游戏
玩家3已加入游戏
玩家5已加入游戏
|
三、源码分析
CyclicBarrier 基于 ReentrantLock 和 Condition 机制实现。除了 getParties() 方法，CyclicBarrier 的其他方法都需要获取锁。
3.1 属性
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| public class CyclicBarrier {
private static class Generation {
boolean broken = false;
}
private final ReentrantLock lock = new ReentrantLock();
private final Condition trip = lock.newCondition();
private final int parties;
private final Runnable barrierCommand;
private Generation generation = new Generation();
private int count;
...
}
|
3.2 构造方法
1
2
3
4
5
6
7
8
9
10
| public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
this(parties, null);
}
|
CyclicBarrier 提供了以上两个构造方法调用：
- 默认的构造方法是 CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用 await 方法告诉 CyclicBarrier 已经到达屏障位置，线程被阻塞。
- 另外一个构造方法 CyclicBarrier(int parties, Runnable barrierAction)，其中 barrierAction 任务会在所有线程到达屏障后执行，方便处理更复杂的业务场景。
3.3 await()
调用 await() 方法的线程告诉 CyclicBarrier 已经到达同步点，然后当前线程被阻塞。直到 parties（设置的屏障数量）个参与线程调用了await方法。CyclicBarrier 同样提供带超时时间的 await() 方法。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe);
}
}
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
|
await() 和 await(long, TimeUnit) 都是调用 dowait() 方法，这个方法是核心方法，在等待线程数未达到 parties 时静静地等待，满足一下条件就会中断等待：
- 最后一个线程到达，即 index == 0
- 某个参与线程等待超时
- 某个参与线程被中断
- 调用了 CyclicBarrier 的 reset() 方法，该方法会将屏障重置为初始状态
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
| private int dowait(boolean timed, long nanos) throws InterruptedException,
BrokenBarrierException, TimeoutException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
final Generation g = generation;
if (g.broken)
throw new BrokenBarrierException();
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
int index = --count;
if (index == 0) {
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
for (;;) {
try {
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
|
列下上面调用的两个方法：
nextGeneration() 在屏障开闸之后重置状态，以待下一次调用。breakBarrier() 在屏障打破之后设定打破状态，以唤醒其他线程并通知。
1
2
3
4
5
6
7
8
9
10
11
| private void nextGeneration() {
trip.signalAll();
count = parties;
generation = new Generation();
}
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}
|
3.4 reset()
reset() 方法将屏障重置为其初始状态，如果所有参与者目前都在屏障处等待，则它们将返回，同时抛出一个BrokenBarrierException。
这里要注意一下要先 broken 当前屏蔽，然后再创建一个新的屏蔽，否则的话可能会导致信号丢失。
1
2
3
4
5
6
7
8
9
10
| public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier();
nextGeneration();
} finally {
lock.unlock();
}
}
|
四、CountDownLatch 的区别
- CountDownLatch 一个线程(或者多个)，等待另外 N 个线程完成某个事情之后才能执行；CyclicBarrier N个线程相互等待，任何一个线程完成之前，所有的线程都必须等待。
- CountDownLatch 是一次性的；CyclicBarrier 可以重复使用。
- CountDownLatch 基于 AQS 技术；CyclicBarrier 基于锁和 Condition。本质上都是依赖于 volatile 和 CAS 实现的。
