CycleBarrier与CountDownLatch原理

CountDownLatch

众所周知，它能解决一个任务必须在其他任务完成的情况下才能执行的问题，代码层面来说就是只有计数countDown到0的时候，await处的代码才能继续向下运行，例如：

1import java.util.*;
2import java.util.concurrent.*;
3
4public class Main {
5    public static void main(String[] args) throws Exception {
6
7        CountDownLatch latch = new CountDownLatch(3);
8
9        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5));
10        Future<Integer>[] futures = new Future[3];
11        for (int i = 0; i < 3; i++){
12            futures[i] = executor.submit(() -> {
13                Random rand = new Random();
14                int n = rand.nextInt(100);
15                int result = 0;
16                for (int j = 0; j < n; j++){
17                    result += j;
18                }
19                System.out.println(result + "|" + Thread.currentThread().getName());
20                latch.countDown();
21                return result;
22            });
23        }
24        latch.await();
25        System.out.println("合计每个任务的结果：" + (futures[0].get()+futures[1].get()+futures[2].get()));
26    }
27
28}

运行结果：

源码

实际上内部十分简单，里面只有一个AQS的子类

1private static final class Sync extends AbstractQueuedSynchronizer {
2    private static final long serialVersionUID = 4982264981922014374L;
3
4    // 它把AQS的state（同步状态）作为计数器，在AQS里，state是个volatile标记的int变量
5    Sync(int count) {
6        setState(count);
7    }
8
9    int getCount() {
10        return getState();
11    }
12
13    protected int tryAcquireShared(int acquires) {
14        // 同步状态为0，则返回1，否则返回-1
15        return (getState() == 0) ? 1 : -1;
16    }
17
18    protected boolean tryReleaseShared(int releases) {
19        // Decrement count; signal when transition to zero
20        for (;;) {
21            int c = getState();
22            // 如果状态为0则返回false
23            if (c == 0)
24                return false;
25            // 计数器减1
26            int nextc = c-1;
27            // CAS操作，如果内存中的同步状态值等于期望值c，那么将同步状态设置为给定的更新值nextc
28            if (compareAndSetState(c, nextc))
29                return nextc == 0;  // 当计数器减到0，返回true
30        }
31    }
32}
33
34public void countDown() {
35    sync.releaseShared(1);
36}
37
38public void await() throws InterruptedException {
39    sync.acquireSharedInterruptibly(1);
40}

下面看具体做了什么事情

先来看await

1public final void acquireSharedInterruptibly(int arg)
2        throws InterruptedException {
3    if (Thread.interrupted())
4        throw new InterruptedException();
5    // 当计数器不等于0，返回-1，证明还有任务未执行完，进入下面方法等待
6    if (tryAcquireShared(arg) < 0)
7        doAcquireSharedInterruptibly(arg);
8}
9
10private void doAcquireSharedInterruptibly(int arg)
11    throws InterruptedException {
12    // 把当前线程包装成Node放入等待队列
13    final Node node = addWaiter(Node.SHARED);
14    boolean failed = true;
15    try {
16        for (;;) {
17            // 获取当前线程的前驱节点，以检查等待状态
18            final Node p = node.predecessor();
19            if (p == head) {
20                // 如果计数器等于0，返回1，证明此时阻塞可以解除了
21                int r = tryAcquireShared(arg);
22                if (r >= 0) {
23                    setHeadAndPropagate(node, r);
24                    p.next = null; // help GC
25                    failed = false;
26                    return;
27                }
28            }
29            if (shouldParkAfterFailedAcquire(p, node) &&
30                parkAndCheckInterrupt())
31                throw new InterruptedException();
32        }
33    } finally {
34        if (failed)
35            cancelAcquire(node);
36    }
37}

上面的过程可以总结为：当进入await方法后，如果此时计数器不为0，则进入死循环一直检查计数器的值，直到为0退出，此时停止等待。

再来看countDown

1public final boolean releaseShared(int arg) {
2    // 尝试计数器减1，只有减到0才会返回true
3    if (tryReleaseShared(arg)) {
4        doReleaseShared();
5        return true;
6    }
7    return false;
8}
9
10private void doReleaseShared() {
11    for (;;) {
12        Node h = head;
13        if (h != null && h != tail) {
14            int ws = h.waitStatus;
15            // 等待状态为SIGNAL
16            if (ws == Node.SIGNAL) {
17                // 把当前节点的等待状态从SIGNAL设置成0，如果设置失败则继续循环。
18                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
19                    continue;            // loop to recheck cases
20                // 成功的话则卸载当前节点的所有后继
21                unparkSuccessor(h);
22            }
23            // 如果等待状态为0，则尝试将状态设置为PROPAGATE，如果设置失败则继续循环。
24            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
25                continue;                // loop on failed CAS
26        }
27        if (h == head)                   // loop if head changed
28            break;
29    }
30}

countDown的过程可以总结为：尝试将计数器-1，直到为0，为0的时候通知等待线程。

CycleBarrier

栏栅的作用就是让指定的一批任务能够同时开始执行，比如

1import java.util.*;
2import java.util.concurrent.*;
3
4public class Main {
5    public static void main(String[] args) throws Exception {
6        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
7
8
9        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5));
10        Future<Integer>[] futures = new Future[3];
11        for (int i = 0; i < 3; i++){
12            futures[i] = executor.submit(() -> {
13                System.out.println("await|" + Thread.currentThread().getName());
14                cyclicBarrier.await();
15                Random rand = new Random();
16                int n = rand.nextInt(100);
17                int result = 0;
18                for (int j = 0; j < n; j++){
19                    result += j;
20                }
21                System.out.println(result + "|" + Thread.currentThread().getName());
22                return result;
23            });
24        }
25    }
26
27}

运行结果

源码

进来之后首先发现的是成员变量

1/** 用来保护栅栏入口的锁 */
2private final ReentrantLock lock = new ReentrantLock();
3/** 等待条件，直到计数器为0 */
4private final Condition trip = lock.newCondition();
5/** 参与线程的个数 */
6private final int parties;
7/* 计数器为0时要运行的命令，由用户定义 */
8private final Runnable barrierCommand;
9/** 当前等待的一代 */
10private Generation generation = new Generation();
11/**
12 * parties数量的等待线程。每一代等待的数量从parties到0。当调用nextGeneration或者breakBarrier方法时重置。
13 */
14private int count;

从这里可以看出，除了内部实现用的ReentrantLock，其工作过程无非：计数器不为0的时候线程等待；当等待线程全部就绪，也就是计数器减为0的时候重置计数器并通知所有线程继续运行。

导致计数器重置原因有两个：一个就是发生异常，将当前这一代标记为无效（broken=true）；另一个就是正常就绪，开启下一代（new Generation）

核心方法dowait

1// 情况一：timed=false，nanos=0L，代表一直阻塞
2// 情况二：timed=true，nanos!=0L，代表在超时时间内阻塞
3private int dowait(boolean timed, long nanos)
4    throws InterruptedException, BrokenBarrierException,
5           TimeoutException {
6    final ReentrantLock lock = this.lock;
7    lock.lock();
8    try {
9        // 获取当前这一代
10        final Generation g = generation;
11
12        // 如果当前这一代已经销毁，抛异常
13        if (g.broken)
14            throw new BrokenBarrierException();
15        // 测试当前线程是否被中断
16        if (Thread.interrupted()) {
17            // 将broken设置为true，代表这一代已经销毁，重置count；然后通知所有等待线程
18            breakBarrier();
19            throw new InterruptedException();
20        }
21        // count 减1
22        int index = --count;
23        // 如果减1之后变成0，证明等待线程全部就绪。
24        if (index == 0) {  // tripped
25            boolean ranAction = false;
26            try {
27                // 如果用户定义了额外的命令，则执行
28                final Runnable command = barrierCommand;
29                if (command != null)
30                    command.run();
31                ranAction = true;
32                // 开启下一代（通知所有等待线程，重置count，new一个新的Generation）
33                nextGeneration();
34                return 0;
35            } finally {
36                if (!ranAction)
37                    breakBarrier();
38            }
39        }
40
41        // loop until tripped, broken, interrupted, or timed out
42        // 如果减1之后不等于0，也就是还有其它线程没有就绪，那么进入此循环，直到就绪或者被销毁，或者被中断和超时
43        for (;;) {
44            try {
45                if (!timed)
46                    // 未定义超时，则一直阻塞
47                    trip.await();
48                else if (nanos > 0L)
49                    // 等待指定的超时时间
50                    nanos = trip.awaitNanos(nanos);
51            } catch (InterruptedException ie) {
52                if (g == generation && ! g.broken) {
53                    breakBarrier();
54                    throw ie;
55                } else {
56                    // We're about to finish waiting even if we had not
57                    // been interrupted, so this interrupt is deemed to
58                    // "belong" to subsequent execution.
59                    Thread.currentThread().interrupt();
60                }
61            }
62
63            if (g.broken)
64                throw new BrokenBarrierException();
65
66            if (g != generation)
67                return index;
68
69            // 超时，则销毁这一代，通知所有等待线程并重置count
70            if (timed && nanos <= 0L) {
71                breakBarrier();
72                throw new TimeoutException();
73            }
74        }
75    } finally {
76        lock.unlock();
77    }
78}

总结

两个工具实现思路都很简单，唯一我思考的是，为什么CountDownLatch只能用一次？

CycleBarrier很明显，它无论正常执行或者发生异常中断都有重置count的逻辑。

而CountDownLatch则没有重置的逻辑，那么，到底是CountDownLatch不能重置还是仅仅因为没有重置的逻辑。为此我把CountDownLatch的代码照搬，然后加上了简单的重置方法，如下：

1import java.util.concurrent.TimeUnit;
2import java.util.concurrent.locks.AbstractQueuedSynchronizer;
3
4
5public class MyCountDown {
6
7    private static final class Sync extends AbstractQueuedSynchronizer {
8        private static final long serialVersionUID = 4982264981922014374L;
9
10        Sync(int count) {
11            setState(count);
12        }
13
14        /**
15         * 新加
16         * @param count
17         */
18        void reset(int count){
19            // 重新设置状态
20            setState(count);
21        }
22
23        int getCount() {
24            return getState();
25        }
26
27        protected int tryAcquireShared(int acquires) {
28            return (getState() == 0) ? 1 : -1;
29        }
30
31        protected boolean tryReleaseShared(int releases) {
32            // Decrement count; signal when transition to zero
33            for (;;) {
34                int c = getState();
35                if (c == 0)
36                    return false;
37                int nextc = c-1;
38                if (compareAndSetState(c, nextc))
39                    return nextc == 0;
40            }
41        }
42    }
43
44    private final Sync sync;
45
46    private final int count;
47
48    public MyCountDown(int count) {
49        if (count < 0) throw new IllegalArgumentException("count < 0");
50        this.sync = new Sync(count);
51        this.count = count;
52    }
53
54
55    public void await() throws InterruptedException {
56        sync.acquireSharedInterruptibly(1);
57    }
58
59    public boolean await(long timeout, TimeUnit unit)
60            throws InterruptedException {
61        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
62    }
63
64
65    public void countDown() {
66        sync.releaseShared(1);
67    }
68
69
70    public long getCount() {
71        return sync.getCount();
72    }
73
74    public String toString() {
75        return super.toString() + "[Count = " + sync.getCount() + "]";
76    }
77
78    /**
79     * 新加
80     */
81    public void reset(){
82        // 调用重置的方法
83        this.sync.reset(count);
84    }
85}

测试：

1import java.util.*;
2import java.util.concurrent.*;
3
4public class Main {
5    public static void main(String[] args) throws Exception {
6
7        MyCountDown myCountDown = new MyCountDown(3);
8        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5));
9        Future<Integer>[] futures = new Future[3];
10        for (int i = 0; i < 3; i++){
11            futures[i] = executor.submit(() -> {
12                Random rand = new Random();
13                int n = rand.nextInt(100);
14                int result = 0;
15                for (int j = 0; j < n; j++){
16                    result += j;
17                }
18                System.out.println(result + "|" + Thread.currentThread().getName());
19                Thread.sleep(new Random().nextInt(2000));   // 模拟耗时
20                myCountDown.countDown();
21                return result;
22            });
23        }
24        myCountDown.await();
25        System.out.println("第一次：" + (futures[0].get() + futures[1].get() + futures[2].get()));
26        myCountDown.reset();    // 重置
27
28        for (int i = 0; i < 3; i++){
29            futures[i] = executor.submit(() -> {
30                Random rand = new Random();
31                int n = rand.nextInt(100);
32                int result = 0;
33                for (int j = 0; j < n; j++){
34                    result += j;
35                }
36                System.out.println(result + "|" + Thread.currentThread().getName());
37                Thread.sleep(new Random().nextInt(2000));   // 模拟耗时
38                myCountDown.countDown();
39                return result;
40            });
41        }
42        myCountDown.await();
43        System.out.println("如果重置无效，则这个信息会先于任务信息输出");
44        System.out.println("第二次：" + (futures[0].get() + futures[1].get() + futures[2].get()));
45    }
46
47}

输出

如果换成CountDownLatch

1import java.util.*;
2import java.util.concurrent.*;
3
4public class Main {
5    public static void main(String[] args) throws Exception {
6
7        CountDownLatch latch = new CountDownLatch(3);
8        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5));
9        Future<Integer>[] futures = new Future[3];
10        for (int i = 0; i < 3; i++){
11            futures[i] = executor.submit(() -> {
12                Random rand = new Random();
13                int n = rand.nextInt(100);
14                int result = 0;
15                for (int j = 0; j < n; j++){
16                    result += j;
17                }
18                System.out.println(result + "|" + Thread.currentThread().getName());
19                Thread.sleep(new Random().nextInt(2000));   // 模拟耗时
20                latch.countDown();
21                return result;
22            });
23        }
24        latch.await();
25        System.out.println("第一次：" + (futures[0].get() + futures[1].get() + futures[2].get()));
26
27        for (int i = 0; i < 3; i++){
28            futures[i] = executor.submit(() -> {
29                Random rand = new Random();
30                int n = rand.nextInt(100);
31                int result = 0;
32                for (int j = 0; j < n; j++){
33                    result += j;
34                }
35                System.out.println(result + "|" + Thread.currentThread().getName());
36                Thread.sleep(new Random().nextInt(2000));   // 模拟耗时
37                latch.countDown();
38                return result;
39            });
40        }
41        latch.await();
42        System.out.println("如果重置无效，则这个信息会先于任务信息输出");
43        System.out.println("第二次：" + (futures[0].get() + futures[1].get() + futures[2].get()));
44    }
45
46}

输出

 所以可以得出结论，CountDownLatch不是没有办法重置，只不过没有写相关逻辑。当然这个问题如果我说错了，望指正。