并发工具CyclicBarrier源码分析及应用

一.CyclicBarrier简介

1.简介

CyclicBarrier是一个同步的辅助类，允许一组线程相互之间等待，达到一个共同点，再继续执行。 CyclicBarrier（循环屏障）直译为可循环使用（Cyclic）的屏障（Barrier）。它可以让一组线程到达一个屏障（同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续工作。

JDK中的描述：

A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. The barrier is called cyclic because it can be re-used after the waiting threads are released. CyclicBarrier是一个同步辅助类，它允许一组线程相互等待直到所有线程都到达一个公共的屏障点。在程序中有固定数量的线程，这些线程有时候必须等待彼此，这种情况下，使用CyclicBarrier很有帮助。这个屏障之所以用循环修饰，是因为在所有的线程释放彼此之后，这个屏障是可以重新使用的

2.运行机制

二.CyclicBarrier结构图

三.CyclicBarrier方法说明

1.CyclicBarrier(parties)

初始化相互等待的线程数量的构造方法

2.CyclicBarrier(parties,RunnablebarrierAction)

初始化相互等待的线程数量的构造方法以及屏障线程的构造方法屏障线程的运行时机：等待的线程数量 = parties，CyclicBarrier打开屏障之前举例：在分组计算中，每个线程负责一部分计算，最终这些线程计算结束之后，交由屏障线程进行汇总计算

3、getParties（）

获取CyclicBarrier打开屏障的线程数量，也成为方数

4、getNumberWaiting()

获取真在CyclicBarrier上等待的线程数量

5、await()

在CyclicBarrier上进行阻塞等待，直到发生以下情形之一：

a.在CyclicBarrier上等待的线程数量达到parties，则所有线程被释放，继续执行。

b.当前线程被中断，则抛出InterruptedException异常，并停止等待，继续执行。

c.其他等待的线程被中断，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。

d.其他等待的线程超时，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。

e.其他线程调用CyclicBarrier.reset()方法，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。

6、await(timeout,TimeUnit)

在CyclicBarrier上进行限时的阻塞等待，直到发生以下情形之一：

a.在CyclicBarrier上等待的线程数量达到parties，则所有线程被释放，继续执行。

b.当前线程被中断，则抛出InterruptedException异常，并停止等待，继续执行。

c.当前线程等待超时，则抛出TimeoutException异常，并停止等待，继续执行。

d.其他等待的线程被中断，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。

e.其他等待的线程超时，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。

f.其他线程调用CyclicBarrier.reset()方法，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。

7、isBroken

获取是否破损标志位broken的值，此值有以下几种情况：

a.CyclicBarrier初始化时，broken=false，表示屏障未破损。

b.如果正在等待的线程被中断，则broken=true，表示屏障破损。

c.如果正在等待的线程超时，则broken=true，表示屏障破损。

d.如果有线程调用CyclicBarrier.reset()方法，则broken=false，表示屏障回到未破损状态。

8、reset

使得CyclicBarrier回归初始状态，直观来看它做了两件事： a.如果有正在等待的线程，则会抛出BrokenBarrierException异常，且这些线程停止等待，继续执行。 b.将是否破损标志位broken置为false。

四.源码分析

首先看一下CyclicBarrier内部声明的一些属性

/**用于保护屏障入口的锁*/

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

/**实际中仍在等待的线程数，每当有一个线程到达屏障点，

count值就会减一；当一次新的运算开始后，

count的值被重置为parties*/

private int count;

其中，Generation是CyclicBarrier的一个静态内部类

它只有一个boolean类型的属性，具体代码如下：

private static class Generation { Generation() {} // prevent access constructor creation boolean broken; // initially false }

当使用构造方法创建CyclicBarrier实例的时候

就是给上面这些属性赋值

//创建一个CyclicBarrier实例，parties指定参与相互等待的线程数，

//barrierAction指定当所有线程到达屏障点之后，

首先执行的操作，该操作由最后一个进入屏障点线程执行.

public CyclicBarrier(int parties,

Runnable barrierAction) {

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

this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }

//创建一个CyclicBarrier实例，parties指定

参与相互等待的线程数

public CyclicBarrier(int parties) { this(parties, null); }

CyclicBarrier.await方法调用CyclicBarrier.dowait()

每次调用await()都会使计数器-1，当减少到0 时就会

唤醒所有的线程，当调用await()方法时，当前线程已经

到达屏障点，当前线程阻塞进入休眠状态

//该方法被调用时表示当前线程已经到达屏障点，

当前线程阻塞进入休眠状态

//直到所有线程都到达屏障点，当前线程才会被唤醒 public int await() throws InterruptedException, BrokenBarrierException { try {return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } }

当前线程已经到达屏障点，当前线程阻塞进入休眠状态

//该方法被调用时表示当前线程已经到达屏障点.当前线程阻塞进入休眠状态

//在timeout指定的超时时间内，等待其他参与线程

到达屏障点

//如果超出指定的等待时间则抛TimeoutException

异常，如果该时间小于等于零，则此方法根本不会等待

public int await(long timeout, TimeUnit unit)

throws InterruptedException, BrokenBarrierException,

TimeoutException { return dowait(true, unit.toNanos(timeout)); }

dowait（）方法

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

//如果当前线程被中断会做以下三件事

//1.打翻当前栅栏

//2.唤醒拦截的所有线程

//3.抛出中断异常

breakBarrier();

throw new InterruptedException(); }

//每调用一次await()方法，计数器就减一

int index = --count;

//计数器的值减为0则需唤醒所有线程并转换到下一代

if (index == 0) { // tripped boolean ranAction = false;

try {

//如果在创建CyclicBarrier实例时设置了barrierAction，则先执行 barrierAction

final Runnable command = barrierCommand;

if (command != null)

command.run();

ranAction = true;

//当所有参与的线程都到达屏障点，为唤醒所有处于

休眠状态的线程做准备工作

//需要注意的是，唤醒所有阻塞线程不是在这里 nextGeneration();

return 0;

} finally {

//确保在任务未成功执行时能将所有线程唤醒

if (!ranAction)

breakBarrier();

}

}

// loop until tripped, broken,

interrupted, or timed out

//如果计数器不为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;

//如果线程因为时间到了而被唤醒则打翻栅栏

并抛出异常

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

breakBarrier();

throw new TimeoutException(); }

}

}

finally { lock.unlock(); } }

每次调用await方法都会使内部的计数器临时变量-1

当减少到0时，就会调用nextGeneration方法

private void nextGeneration() {

// signal completion of last generation trip.signalAll();

// set up next generation

count = parties; generation = new Generation(); }

在这里唤醒所有阻塞的线程

提醒：在声明CyclicBarrier的时候还可以传一个Runnable的实现类，当计数器减少到0时，

会执行该实现类到这里CyclicBarrier的

实现原理基本已经都清楚了下面来深入源码分析

一下线程阻塞代码 trip.await()和线程唤醒trip.signalAll()的实现。

//await()是AQS内部类ConditionObject中的方法 public final void await() throws InterruptedException {

//如果线程中断抛异常

if (Thread.interrupted())

throw new InterruptedException();

//新建Node节点，并将新节点加入到Condition

等待队列中

//Condition等待队列是AQS内部类

ConditionObject实现的，ConditionObject

有两个属性，

分别是firstWaiter和lastWaiter

都是Node类型

//firstWaiter和lastWaiter分别用于代表Condition等待队列的头结点和尾节点 Node node = addConditionWaiter();

//释放独占锁，让其它线程可以获取到dowait()

方法中的独占锁

int savedState = fullyRelease(node);

int interruptMode = 0;

//检测此节点是否在资源等待队列(AQS同步队列)中， //如果不在，说明此线程还没有竞争资源锁的权利，

此线程继续阻塞，直到检测到此节点在资源等待队列上(AQS同步队列)中

//这里出现了两个等待队列，分别是Condition等待

队列和AQS资源锁等待队列(或者说是同步队列) //Condition等待队列是等待被唤醒的线程队列

AQS资源锁等待队列是等待获取资源锁的队列

while (!isOnSyncQueue(node))

{

//阻塞当前线程，当前线程进入休眠状态，可以看到

这里使用LockSupport.park阻塞当前线程 LockSupport.park(this);

if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; }

if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode

= REINTERRUPT;

if (node.nextWaiter != null)

// clean up if cancelled unlinkCancelledWaiters();

if (interruptMode != 0) reportInterruptAfterWait(interruptMode); }

//addConditionWaiter()是AQS内部类ConditionObject中的方法

private Node addConditionWaiter()

{ Node t = lastWaiter;

// 将condition等待队列中，节点状态不是

CONDITION的节点，从condition等待队列中移除

if (t != null && t.waitStatus != Node.CONDITION)

{ unlinkCancelledWaiters();

t = lastWaiter; }

//以下操作是用此线程构造一个节点，并将之加入到condition等待队列尾部

Node node = new Node(Thread.currentThread(), Node.CONDITION);

if (t == null)

firstWaiter = node;

else

t.nextWaiter = node;

lastWaiter = node; return node; }

//signalAll是AQS内部类ConditionObject中

的方法

public final void signalAll() {

if (!isHeldExclusively())

throw new IllegalMonitorStateException(); //Condition等待队列的头结点

Node first = firstWaiter;

if (first != null) doSignalAll(first); }

private void doSignalAll(Node first) { lastWaiter = firstWaiter = null; do {Node next = first.nextWaiter; first.nextWaiter = null;

//将Condition等待队列中的Node节点按之前顺序

都转移到了AQS同步队列中 transferForSignal(first);

first = next;

} while (first != null); }

final boolean transferForSignal

(Node node) {

if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))

return false;

//这里将Condition等待队列中的Node节点插入到

AQS同步队列的尾部 Node p = enq(node);

int ws = p.waitStatus; if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread); return true; }

//ReentrantLock#unlock()方法

public void unlock() {

//Sync是ReentrantLock的内部类，继承自AbstractQueuedSynchronizer，

它是 ReentrantLock中公平锁和非公平锁的基础实现 sync.release(1); }

public final boolean release(int arg) {

//释放锁

if (tryRelease(arg)) {

//AQS同步队列头结点

Node h = head; if (h != null && h.waitStatus != 0)

//唤醒节点中的线程

unparkSuccessor(h); return true; }

return false; }

private void unparkSuccessor(Node node)

{ int ws = node.waitStatus;

if (ws < 0) compareAndSetWaitStatus(node, ws, 0);

Node s = node.next;

if (s == null || s.waitStatus > 0)

{ s = null; for (Node t = tail;

t != null && t != node; t = t.prev)

if (t.waitStatus <= 0) s = t; }

if (s != null) //唤醒阻塞线程 LockSupport.unpark(s.thread); }

365天干货不断,可以微信搜索「猿灯塔」第一时间阅读，回复【资料】【面试】【简历】有我准备的一线大厂面试资料和简历模板