转载

JDK源码分析-CyclicBarrier

概述

CyclicBarrier 是并发包中的一个工具类,它的典型应用场景为: 几个线程执行完任务后,执行另一个线程(回调函数,可选), 然后继续下一轮,如此往复。

打个通俗的比方,可以把 CyclicBarrier 的执行流程比作:几个人(类比线程)围着操场跑圈,所有人都到达终点后(终点可理解为“屏障(barrier)”,到达次序可能有先后,对应线程执行任务有快慢),执行某个操作(回调函数),然后再继续跑下一圈(下一次循环),如此往复。

该类与 CountDownLatch 相比,可以把后者理解为“一次性(one-shot)”操作,而前者是“可循环”的操作,下面分析其代码实现。

代码分析

CyclicBarrier 的主要方法如下:

JDK源码分析-CyclicBarrier

其中常用的是两个 await 方法,作用是让当前线程进入等待状态。

成员变量及嵌套类:


 

// 内部嵌套类

private static class Generation {

boolean broken = false;

}


/** The lock for guarding barrier entry */

private final ReentrantLock lock = new ReentrantLock();


/** Condition to wait on until tripped */

private final Condition trip = lock.newCondition();


/** The number of parties */

private final int parties;


/* The command to run when tripped */

private final Runnable barrierCommand;


/** The current generation */

private Generation generation = new Generation();


/**

* Number of parties still waiting. Counts down from parties to 0

* on each generation. It is reset to parties on each new

* generation or when broken.

*/

private int count;

内部嵌套类 Generation 表示代数,每次屏障(barrier)破坏之前属于同一代,之后进入下一代。

构造器:


 

// 无回调函数

public CyclicBarrier(int parties) {

this(parties, null);

}


// 有回调函数

public CyclicBarrier(int parties, Runnable barrierAction) {

if (parties <= 0) throw new IllegalArgumentException();

this.parties = parties;

this.count = parties;

this.barrierCommand = barrierAction;

}

CyclicBarrier 有两个构造器,其中后者可以传入一个回调函数(barrierAction),parties 表示调用 await 的线程数。

await 方法:


 

// 阻塞式等待

public int await() throws InterruptedException, BrokenBarrierException {

try {

return dowait(false, 0L);

} catch (TimeoutException toe) {

throw new Error(toe); // cannot happen

}

}


// 有超时的等待

public int await(long timeout, TimeUnit unit)

throws InterruptedException,

BrokenBarrierException,

TimeoutException {

return dowait(true, unit.toNanos(timeout));

}

可以看到两个 await 方法都是调用 dowait 方法来实现的(该方法也是 CyclicBarrier 的核心方法),如下:


 

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();

}


// count 减 1

int index = --count;

if (index == 0) { // tripped

// count 减到 0 时触发的操作

boolean ranAction = false;

try {

// 传入的回调函数

final Runnable command = barrierCommand;

if (command != null)

// 若传了回调函数,则执行回调函数

// PS: 由此可知,回调函数由最后一个执行结束的线程执行

command.run();

ranAction = true;

// 进入下一代(下一轮操作)

nextGeneration();

return 0;

} finally {

if (!ranAction)

breakBarrier();

}

}


// loop until tripped, broken, interrupted, or timed out

for (;;) {

try {

// count 不为 0 时,当前线程进入等待状态

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;

if (timed && nanos <= 0L) {

breakBarrier();

throw new TimeoutException();

}

}

} finally {

lock.unlock();

}

}

nextGeneration 和 breakBarrier:


 

// 进入下一轮

private void nextGeneration() {

// signal completion of last generation

trip.signalAll();

// set up next generation

count = parties;

generation = new Generation();

}


// 破坏屏障

private void breakBarrier() {

generation.broken = true;

count = parties;

trip.signalAll();

}

执行流程: 初始化时 parties 和 count 的值相同(由构造器 parties 参数传入),之后每有一个线程调用 await 方法 count 值就减 1,直至 count 为 0 时(若不为 0 则等待),执行传入的回调函数 barrierCommand(若不为空),然后唤醒所有线程,并将 count 重置为 parties,开始下一轮操作。

场景举例

为了便于理解 CyclicBarrier 的用法,下面简单举例演示(仅供参考):


 

public class CyclicBarrierTest {

private static final int COUNT = 3;


public static void main(String[] args) throws InterruptedException {

// 初始化 CyclicBarrier 对象及回调函数

CyclicBarrier cyclicBarrier = new CyclicBarrier(COUNT, () -> {

// 模拟回调函数的操作(模拟写操作)

System.out.println(Thread.currentThread().getName() + " start writing..");

try {

TimeUnit.SECONDS.sleep(1);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("---------");

});


while (true) {

// 创建几个线程执行任务

for (int i = 0; i < COUNT; i++) {

new Thread(() -> {

// 模拟读操作

System.out.println(Thread.currentThread().getName() + " is reading..");

try {

TimeUnit.SECONDS.sleep(3);

// 等待

cyclicBarrier.await();

} catch (InterruptedException | BrokenBarrierException e) {

e.printStackTrace();

}

}).start();

}

// 睡眠 10 秒,然后进入下一轮

TimeUnit.SECONDS.sleep(10);

}

}

}


/* 执行结果(仅供参考):

Thread-0 is reading..

Thread-1 is reading..

Thread-2 is reading..

Thread-1 start writing..

---------

Thread-3 is reading..

Thread-4 is reading..

Thread-5 is reading..

Thread-5 start writing..

---------

*/

PS: 此处模拟多个线程执行读操作,都读完后再执行写操作;之后再读、再写……可以理解为简单的对账系统。

此处代码仅供参考,只为便于理解该类的用法。实际上每次创建线程是不合理的(可以使用线程池,由于未分析,这里暂不使用)。

小结

CyclicBarrier 也可以理解为倒数的计数器, 它与 CountDownLatch 有些类似。后者是“一次性”的,而前者是“可循环使用”的

Stay hungry, stay foolish.

JDK源码分析-CyclicBarrier

原文  http://mp.weixin.qq.com/s?__biz=MzU4NzYyMDE4MQ==&mid=2247483932&idx=1&sn=ab8ea3d143f989f7edb69bc6552df294
正文到此结束
Loading...