CycleBarrier与CountDownLatch原理
Posted lua123
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了CycleBarrier与CountDownLatch原理相关的知识,希望对你有一定的参考价值。
CountDownLatch
众所周知,它能解决一个任务必须在其他任务完成的情况下才能执行的问题,代码层面来说就是只有计数countDown到0的时候,await处的代码才能继续向下运行,例如:
import java.util.*; import java.util.concurrent.*; public class Main { public static void main(String[] args) throws Exception { CountDownLatch latch = new CountDownLatch(3); ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5)); Future<Integer>[] futures = new Future[3]; for (int i = 0; i < 3; i++){ futures[i] = executor.submit(() -> { Random rand = new Random(); int n = rand.nextInt(100); int result = 0; for (int j = 0; j < n; j++){ result += j; } System.out.println(result + "|" + Thread.currentThread().getName()); latch.countDown(); return result; }); } latch.await(); System.out.println("合计每个任务的结果:" + (futures[0].get()+futures[1].get()+futures[2].get())); } }
运行结果:
源码
实际上内部十分简单,里面只有一个AQS的子类
private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; // 它把AQS的state(同步状态)作为计数器,在AQS里,state是个volatile标记的int变量 Sync(int count) { setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { // 同步状态为0,则返回1,否则返回-1 return (getState() == 0) ? 1 : -1; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); // 如果状态为0则返回false if (c == 0) return false; // 计数器减1 int nextc = c-1; // CAS操作,如果内存中的同步状态值等于期望值c,那么将同步状态设置为给定的更新值nextc if (compareAndSetState(c, nextc)) return nextc == 0; // 当计数器减到0,返回true } } } public void countDown() { sync.releaseShared(1); } public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); }
下面看具体做了什么事情
先来看await
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 当计数器不等于0,返回-1,证明还有任务未执行完,进入下面方法等待 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); } private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { // 把当前线程包装成Node放入等待队列 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { // 获取当前线程的前驱节点,以检查等待状态 final Node p = node.predecessor(); if (p == head) { // 如果计数器等于0,返回1,证明此时阻塞可以解除了 int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
上面的过程可以总结为:当进入await方法后,如果此时计数器不为0,则进入死循环一直检查计数器的值,直到为0退出,此时停止等待。
再来看countDown
public final boolean releaseShared(int arg) { // 尝试计数器减1,只有减到0才会返回true if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; // 等待状态为SIGNAL if (ws == Node.SIGNAL) { // 把当前节点的等待状态从SIGNAL设置成0,如果设置失败则继续循环。 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases // 成功的话则卸载当前节点的所有后继 unparkSuccessor(h); } // 如果等待状态为0,则尝试将状态设置为PROPAGATE,如果设置失败则继续循环。 else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } }
countDown的过程可以总结为:尝试将计数器-1,直到为0,为0的时候通知等待线程。
CycleBarrier
栏栅的作用就是让指定的一批任务能够同时开始执行,比如
import java.util.*; import java.util.concurrent.*; public class Main { public static void main(String[] args) throws Exception { CyclicBarrier cyclicBarrier = new CyclicBarrier(3); ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5)); Future<Integer>[] futures = new Future[3]; for (int i = 0; i < 3; i++){ futures[i] = executor.submit(() -> { System.out.println("await|" + Thread.currentThread().getName()); cyclicBarrier.await(); Random rand = new Random(); int n = rand.nextInt(100); int result = 0; for (int j = 0; j < n; j++){ result += j; } System.out.println(result + "|" + Thread.currentThread().getName()); return result; }); } } }
运行结果
源码
进来之后首先发现的是成员变量
/** 用来保护栅栏入口的锁 */ private final ReentrantLock lock = new ReentrantLock(); /** 等待条件,直到计数器为0 */ private final Condition trip = lock.newCondition(); /** 参与线程的个数 */ private final int parties; /* 计数器为0时要运行的命令,由用户定义 */ private final Runnable barrierCommand; /** 当前等待的一代 */ private Generation generation = new Generation(); /** * parties数量的等待线程。每一代等待的数量从parties到0。当调用nextGeneration或者breakBarrier方法时重置。 */ private int count;
从这里可以看出,除了内部实现用的ReentrantLock,其工作过程无非:计数器不为0的时候线程等待;当等待线程全部就绪,也就是计数器减为0的时候重置计数器并通知所有线程继续运行。
导致计数器重置原因有两个:一个就是发生异常,将当前这一代标记为无效(broken=true);另一个就是正常就绪,开启下一代(new Generation)
核心方法dowait
// 情况一:timed=false,nanos=0L,代表一直阻塞 // 情况二:timed=true,nanos!=0L,代表在超时时间内阻塞 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()) { // 将broken设置为true,代表这一代已经销毁,重置count;然后通知所有等待线程 breakBarrier(); throw new InterruptedException(); } // count 减1 int index = --count; // 如果减1之后变成0,证明等待线程全部就绪。 if (index == 0) { // tripped boolean ranAction = false; try { // 如果用户定义了额外的命令,则执行 final Runnable command = barrierCommand; if (command != null) command.run(); ranAction = true; // 开启下一代(通知所有等待线程,重置count,new一个新的Generation) nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } } // loop until tripped, broken, interrupted, or timed out // 如果减1之后不等于0,也就是还有其它线程没有就绪,那么进入此循环,直到就绪或者被销毁,或者被中断和超时 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 { // We‘re about to finish waiting even if we had not // been interrupted, so this interrupt is deemed to // "belong" to subsequent execution. Thread.currentThread().interrupt(); } } if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; // 超时,则销毁这一代,通知所有等待线程并重置count if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }
总结
两个工具实现思路都很简单,唯一我思考的是,为什么CountDownLatch只能用一次?
CycleBarrier很明显,它无论正常执行或者发生异常中断都有重置count的逻辑。
而CountDownLatch则没有重置的逻辑,那么,到底是CountDownLatch不能重置还是仅仅因为没有重置的逻辑。为此我把CountDownLatch的代码照搬,然后加上了简单的重置方法,如下:
import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.AbstractQueuedSynchronizer; public class MyCountDown { private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; Sync(int count) { setState(count); } /** * 新加 * @param count */ void reset(int count){ // 重新设置状态 setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } } private final Sync sync; private final int count; public MyCountDown(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); this.count = count; } public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } public void countDown() { sync.releaseShared(1); } public long getCount() { return sync.getCount(); } public String toString() { return super.toString() + "[Count = " + sync.getCount() + "]"; } /** * 新加 */ public void reset(){ // 调用重置的方法 this.sync.reset(count); } }
测试:
import java.util.*; import java.util.concurrent.*; public class Main { public static void main(String[] args) throws Exception { MyCountDown myCountDown = new MyCountDown(3); ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5)); Future<Integer>[] futures = new Future[3]; for (int i = 0; i < 3; i++){ futures[i] = executor.submit(() -> { Random rand = new Random(); int n = rand.nextInt(100); int result = 0; for (int j = 0; j < n; j++){ result += j; } System.out.println(result + "|" + Thread.currentThread().getName()); Thread.sleep(new Random().nextInt(2000)); // 模拟耗时 myCountDown.countDown(); return result; }); } myCountDown.await(); System.out.println("第一次:" + (futures[0].get() + futures[1].get() + futures[2].get())); myCountDown.reset(); // 重置 for (int i = 0; i < 3; i++){ futures[i] = executor.submit(() -> { Random rand = new Random(); int n = rand.nextInt(100); int result = 0; for (int j = 0; j < n; j++){ result += j; } System.out.println(result + "|" + Thread.currentThread().getName()); Thread.sleep(new Random().nextInt(2000)); // 模拟耗时 myCountDown.countDown(); return result; }); } myCountDown.await(); System.out.println("如果重置无效,则这个信息会先于任务信息输出"); System.out.println("第二次:" + (futures[0].get() + futures[1].get() + futures[2].get())); } }
输出
如果换成CountDownLatch
import java.util.*; import java.util.concurrent.*; public class Main { public static void main(String[] args) throws Exception { CountDownLatch latch = new CountDownLatch(3); ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 15, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5)); Future<Integer>[] futures = new Future[3]; for (int i = 0; i < 3; i++){ futures[i] = executor.submit(() -> { Random rand = new Random(); int n = rand.nextInt(100); int result = 0; for (int j = 0; j < n; j++){ result += j; } System.out.println(result + "|" + Thread.currentThread().getName()); Thread.sleep(new Random().nextInt(2000)); // 模拟耗时 latch.countDown(); return result; }); } latch.await(); System.out.println("第一次:" + (futures[0].get() + futures[1].get() + futures[2].get())); for (int i = 0; i < 3; i++){ futures[i] = executor.submit(() -> { Random rand = new Random(); int n = rand.nextInt(100); int result = 0; for (int j = 0; j < n; j++){ result += j; } System.out.println(result + "|" + Thread.currentThread().getName()); Thread.sleep(new Random().nextInt(2000)); // 模拟耗时 latch.countDown(); return result; }); } latch.await(); System.out.println("如果重置无效,则这个信息会先于任务信息输出"); System.out.println("第二次:" + (futures[0].get() + futures[1].get() + futures[2].get())); } }
输出
所以可以得出结论,CountDownLatch不是没有办法重置,只不过没有写相关逻辑。当然这个问题如果我说错了,望指正。
以上是关于CycleBarrier与CountDownLatch原理的主要内容,如果未能解决你的问题,请参考以下文章
Java Review - 并发编程_ 信号量Semaphore原理&源码剖析