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

public class CountDownLatch {private final Sync sync;/** CountDownLatch 的构造方法，该方法要求传入大于0的整型数值作为计数器 */public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");// 初始化 Syncthis.sync = new Sync(count);}/** CountDownLatch 的同步控制器，继承自 AQS */private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = 4982264981922014374L;Sync(int count) {// 设置 AQS statesetState(count);}int getCount() {return getState();}/** 尝试在共享状态下获取同步状态，该方法在 AQS 中是抽象方法，这里进行了覆写 */protected int tryAcquireShared(int acquires) {/** 如果 state = 0，则返回1，表明可获取同步状态，* 此时线程调用 await 方法时就不会被阻塞。*/ return (getState() == 0) ? 1 : -1;}/** 尝试在共享状态下释放同步状态，该方法在 AQS 中也是抽象方法 */protected boolean tryReleaseShared(int releases) {/** 下面的逻辑是将 state--，state 减至0时，调用 await 等待的线程会被唤醒。* 这里使用循环 + CAS，表明会存在竞争的情况，也就是多个线程可能会同时调用 * countDown 方法。在 state 不为0的情况下，线程调用 countDown 是必须要完* 成 state-- 这个操作。所以这里使用了循环 + CAS，确保 countDown 方法可正* 常运行。*/for (;;) {// 获取 stateint c = getState();if (c == 0)return false;int nextc = c-1;// 使用 CAS 设置新的 state 值if (compareAndSetState(c, nextc))return nextc == 0;}}}
}

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

/** * 该方法会使线程进入等待状态，直到计数器减至0，或者线程被中断。当计数器为0时，调用* 此方法将会立即返回，不会被阻塞住。*/
public void await() throws InterruptedException {// 调用 AQS 中的 acquireSharedInterruptibly 方法sync.acquireSharedInterruptibly(1);
}/** 带有超时功能的 await */
public boolean await(long timeout, TimeUnit unit)throws InterruptedException {return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}+--- AbstractQueuedSynchronizer
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {// 若线程被中断，则直接抛出中断异常if (Thread.interrupted())throw new InterruptedException();// 调用 Sync 中覆写的 tryAcquireShared 方法，尝试获取同步状态if (tryAcquireShared(arg) < 0)/** 若 tryAcquireShared 小于0，则表示获取同步状态失败，* 此时将线程放入 AQS 的同步队列中进行等待。*/ doAcquireSharedInterruptibly(arg);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

/** 该方法的作用是将计数器进行自减操作，当计数器为0时，唤醒正在同步队列中等待的线程 */
public void countDown() {// 调用 AQS 中的 releaseShared 方法sync.releaseShared(1);
}+--- AbstractQueuedSynchronizer
public final boolean releaseShared(int arg) {// 调用 Sync 中的 tryReleaseShared 尝试释放同步状态if (tryReleaseShared(arg)) {/** tryReleaseShared 返回 true 时，表明 state = 0，即计数器为0。此时调用* doReleaseShared 方法唤醒正在同步队列中等待的线程*/ doReleaseShared();return true;}return false;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

/** 创建一个允许 parties 个线程通行的屏障 */
public CyclicBarrier(int parties) {this(parties, null);
}/** * 创建一个允许 parties 个线程通行的屏障，若 barrierAction 回调对象不为 null，* 则在最后一个线程到达屏障后，执行相应的回调逻辑*/
public CyclicBarrier(int parties, Runnable barrierAction) {if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;this.count = parties;this.barrierCommand = barrierAction;
}

1
2
3
4
5
6
7
8
9
10
11

/** * CyclicBarrier 是可循环使用的屏障，这里使用 Generation 记录当前轮次 CyclicBarrier * 的运行状态。当所有线程到达屏障后，generation 将会被更新，表示 CyclicBarrier 进入新一* 轮的运行轮次中。
*/
private Generation generation = new Generation();private static class Generation {// 用于记录屏障有没有被破坏boolean broken = false;
}

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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

public int await() throws InterruptedException, BrokenBarrierException {try {// await 的逻辑封装在 dowait 中return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe); // cannot happen}
}private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException, TimeoutException {final ReentrantLock lock = this.lock;// 加锁lock.lock();try {final Generation g = generation;// 如果 g.broken = true，表明屏障被破坏了，这里直接抛出异常if (g.broken)throw new BrokenBarrierException();// 如果线程中断，则调用 breakBarrier 破坏屏障if (Thread.interrupted()) {breakBarrier();throw new InterruptedException();}/** index 表示线程到达屏障的顺序，index = parties - 1 表明当前线程是第一个* 到达屏障的。index = 0，表明当前线程是最有一个到达屏障的。*/ int index = --count;// 当 index = 0 时，唤醒所有处于等待状态的线程if (index == 0) {  // trippedboolean ranAction = false;try {final Runnable command = barrierCommand;// 如果回调对象不为 null，则执行回调if (command != null)command.run();ranAction = true;// 重置屏障状态，使其进入新一轮的运行过程中nextGeneration();return 0;} finally {// 若执行回调的过程中发生异常，此时调用 breakBarrier 破坏屏障if (!ranAction)breakBarrier();}}// 线程运行到此处的线程都会被屏障挡住，并进入等待状态。for (;;) {try {if (!timed)trip.await();else if (nanos > 0L)nanos = trip.awaitNanos(nanos);} catch (InterruptedException ie) {/** 若下面的条件成立，则表明本轮运行还未结束。此时调用 breakBarrier * 破坏屏障，唤醒其他线程，并抛出异常*/ if (g == generation && ! g.broken) {breakBarrier();throw ie;} else {/** 若上面的条件不成立，则有两种可能：* 1. g != generation*     此种情况下，表明循环屏障的第 g 轮次的运行已经结束，屏障已经*     进入了新的一轮运行轮次中。当前线程在稍后返回 到达屏障 的顺序即可*     * 2. g = generation 但 g.broken = true*     此种情况下，表明已经有线程执行过 breakBarrier 方法了，当前*     线程则会在稍后抛出 BrokenBarrierException*/Thread.currentThread().interrupt();}}// 屏障被破坏，则抛出 BrokenBarrierException 异常if (g.broken)throw new BrokenBarrierException();// 屏障进入新的运行轮次，此时返回线程在上一轮次到达屏障的顺序if (g != generation)return index;// 超时判断if (timed && nanos <= 0L) {breakBarrier();throw new TimeoutException();}}} finally {lock.unlock();}
}/** 开启新的一轮运行过程 */
private void nextGeneration() {// 唤醒所有处于等待状态中的线程trip.signalAll();// 重置 countcount = parties;// 重新创建 Generation，表明进入循环屏障进入新的一轮运行轮次中generation = new Generation();
}/** 破坏屏障 */
private void breakBarrier() {// 设置屏障是否被破坏标志generation.broken = true;// 重置 countcount = parties;// 唤醒所有处于等待状态中的线程trip.signalAll();
}

1
2
3
4
5
6
7
8
9
10
11
12

public void reset() {final ReentrantLock lock = this.lock;lock.lock();try {// 破坏屏障breakBarrier();   // break the current generation// 开启新一轮的运行过程nextGeneration(); // start a new generation} finally {lock.unlock();}
}