Java并发编程(十六):CyclicBarrier源码分析
Posted 黄智霖-blog
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java并发编程(十六):CyclicBarrier源码分析相关的知识,希望对你有一定的参考价值。
前言
CyclicBarrier可以建立一个屏障,这个屏障可以阻塞一个线程直到指定的所有线程都达到屏障。就像团队聚餐,等所有人都到齐了再一起动筷子。根据Cyclic就可以发现CyclicBarrier可以重复使用。现在有了前面分析ReentrantLock、Semaphore、CountDownLatch的经验,CyclicBarrier也不复杂了,只是这里又引入新的概念:Condition条件队列,这也是最开始我们分析AQS没有讲的东西。
注:看本文前建议先看看ReentrantLock源码分析、Semaphore源码分析、CountDownLatch源码分析
首先还是来看使用:
CyclicBarrier cb = new CyclicBarrier(10, () ->
System.out.println("所有人员到齐,准备开饭!");
);
ExecutorService es = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++)
es.execute(() ->
try
Thread.sleep(new Random().nextInt(5) * 1000);
System.out.println(Thread.currentThread().getId() + "-已入座,等待其他人...");
cb.await();
System.out.println(Thread.currentThread().getId() + "-开始吃饭");
catch (InterruptedException e)
e.printStackTrace();
catch (BrokenBarrierException e)
e.printStackTrace();
);
es.shutdown();
初始化的时候可以指定一个parties表示等待线程的数量,每当一个线程调用await方法就表示一个线程已经准备好,线程会被阻塞,直到指定数量的线程都准备好才被唤醒。同时提供了一个可选的Runnable参数,当所有线程都准备好之后,唤醒阻塞线程之前会先同步执行这个Runnable。
源码分析
CyclicBarrier的结构没有CountDownLatch和Semaphore那些那样简单的使用一个内部类继承AQS,然后重写几个方法就实现了,其同时依赖了条件队列和同步队列。先来看看CyclicBarrier的类结构:
public class CyclicBarrier
//使用ReentrantLock做同步锁
private final ReentrantLock lock = new ReentrantLock();
//通过lock创建一个Condition,实际上是一个ConditionObject
private final Condition trip = lock.newCondition();
//等待线程的数量,就是构造方法的入参,设置之后不会改变
private final int parties;
//内部维护的计数器,初始状态和parties相同,实际操作的是这个字段
private int count;
//所有线程到达之后执行的任务,可以不指定
private final Runnable barrierCommand;
//分代,CyclicBarrier可以重复使用,可以理解为一轮一轮的,每一轮就是一个Generation
private Generation generation = new Generation();
/*表示分代,内部类*/
private static class Generation
//表示当前分代(轮)是否“中断”
boolean broken = false;
为了达到可以重复使用的目的,CyclicBarrier引入了Generation(分代)的概念,可以将其理解为一轮生命周期,每一轮都有一个Generation,对应到代码中有一个布尔类型的broke字段代表当前轮的生命周期是否被中断,如果被中断会有一系列的处理措施。
首先来看看CyclicBarrier的构造函数:
public CyclicBarrier(int parties)
this(parties, null);
public CyclicBarrier(int parties, Runnable barrierAction)
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
//将parties赋值给count
this.count = parties;
this.barrierCommand = barrierAction;
提供了两个构造函数,初始会将parties赋值给count,然后提供了一个可选的barrierAction参数,会在所有线程准备就绪的时候被同步调用。
接下来看看核心的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方法,该方法有两个入参,表示是否进行超时等待和超时的时间,那么我们就接着进入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())
//线程在lock中阻塞的过程中可能被中断,这里要判断一下中断标识
//如果当前线程被中断过,那么这里手动打断当前代
//会重置count,并且唤醒所有阻塞线程
breakBarrier();
throw new InterruptedException();
//--count,表示当前线程准备就绪
int index = --count;
if (index == 0) // tripped
//表示所有线程准备就绪,需要唤醒阻塞线程
boolean ranAction = false;
try
final Runnable command = barrierCommand;
if (command != null)
//如果指定了command,在这里同步调用
command.run();
ranAction = true;
//开始下一轮
nextGeneration();
return 0;
finally
if (!ranAction)
//说明执行command的时候出现了异常
//需要打断当前代
breakBarrier();
// loop until tripped, broken, interrupted, or timed out
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
//可能的情况就是g != generation || g.broken
//说明已经换代或者已经被打断,这里自我中断向外传递状态
Thread.currentThread().interrupt();
if (g.broken)
//当前代被打断,比如某个线程等待超时被唤醒之后
//会强制打断当前代,抛出TimeoutException,并且唤醒其它阻塞线程
//线程被唤醒后发现当前代被打断,那么这里直接抛出BrokenBarrierException
//或者执行command出现异常
throw new BrokenBarrierException();
if (g != generation)
//说明开始了新的代,当前线程是从同步队列中被唤醒的
//返回index,会在finally块中执行unlock唤醒同步队列中后面的阻塞线程
return index;
if (timed && nanos <= 0L)
//超时唤醒,强制打断当前代,唤醒所有阻塞线程
breakBarrier();
throw new TimeoutException();
finally
//释放锁,这里很重要,线程传递唤醒是在这里处理的
//在finally块中保证传递唤醒不会异常被中断
lock.unlock();
/*手动中断当前代*/
private void breakBarrier()
generation.broken = true;
count = parties;
//唤醒所有阻塞线程
trip.signalAll();
/*换代*/
private void nextGeneration()
//唤醒阻塞线程
trip.signalAll();
//重置count
count = parties;
//换代
generation = new Generation();
可以看到和前面几个工具的CAS操作不同,这里上来就是一个同步锁,之所以使用同步锁,是因为这里不是简单的更新count字段需要保证并发安全,并且线程阻塞唤醒也是依赖同步锁的解锁操作,其中还有较为繁杂的逻辑需要处理。
从这个await方法的逻辑中我们可以看出CyclicBarrier的一个大致工作流程:
- 首先需要通过ReentrantLock加锁
- 一个线程调用了await方法获取到锁,就代表该线程准备就绪,将count减1
- 如果count减一之后大于0,就代表还有线程没有准备就绪,那么需要阻塞当前线程
- 如果count减一之后等于0,就代表所有线程都准备就绪,那么需要同步调用创建CyclicBarrier时指定的command(如果有的话)。然后唤醒阻塞线程,并且进行换代操作,将count重置为初始值(也就是parties),这样CyclicBarrier就能重复使用了
- 如果运行command出现了异常,那么会导致当前代被"中断",仍然会唤醒所有阻塞线程,只是线程被唤醒后发现当前代被打断,那么继而会抛出BrokenBarrierException异常
- 既然调用了ReentrantLock的lock方法,那么不要忘记前面分析ReentrantLock的内容,这里默认创建的是非公平锁,并且lock方法不会抛出中断异常,但是会向外传递中断状态,所以如果线程在lock期间被中断,那么需要在后续逻辑中获取到中断状态,然后手动"中断"当前代,重置count并且唤醒所有阻塞线程
- Condition.await方法可能会抛出中断异常,如果当前代的一个阻塞线程被中断,也会导致重置count,唤醒所有阻塞线程,然后向外抛出异常;如果线程中断的时候已经换代或者当前代已经被"中断",那么只需要自我中断打上中断标记,向外传递即可
- 如果是带超时的等待,一个线程等待超时了,同样会"中断"当前代唤醒阻塞线程,并且抛出超时异常
其实上面根据dowait方法分析出来的逻辑已经大概涵盖了CyclicBarrier的核心内容,但是我们难免还是会有所疑问:
- 阻塞线程存放到哪里的?还是依赖的CLH队列吗?
- 线程是如何阻塞/唤醒的?还是简单的park/unpark吗?
- 条件队列是如何使用的?
- 等等
为了解开这些疑问,我们还需要深入到各个方法调用的细节~ 但是这里先声明一点,源码中有很多关于各种情况下中断唤醒的判断处理(主要是打断分代和抛出异常),而且是结合条件队列和CLH队列的共同使用,细节点也很多,一个方法的编码逻辑考虑的情况可能都够写一个小节,所以本文先从总体脉络上进行梳理。
ReentrantLock的lock方法的中断处理在其源码分析文章中详细说明过,这里我们不多说,首先来看看CyclicBarrier的线程是如何被阻塞的?也就是进入AQS中ConditionObject类的await方法:
public final void await() throws InterruptedException
if (Thread.interrupted())
throw new InterruptedException();
//添加当前线程到条件等待队列
Node node = addConditionWaiter();
//释放AQS中的state,相当于当前线程已经入队,那么可以"释放锁",但是没有执行释放锁的逻辑
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node))
//如果当前节点不再同步队列CLH中,通过park阻塞当前线程
LockSupport.park(this);
//检测唤醒类型,一共有三个状态:0、1、-1
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
//如果不等于0,则跳出循环
break;
//acquireQueued是ReentrantLock入队阻塞的逻辑
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
阻塞的第一步就是addConditionWaiter添加到条件等待队列,这个队列不再是我们熟悉的CLH队列,看看该方法的实现:
private Node addConditionWaiter()
Node t = lastWaiter;
//如果ws不是CONDITION,说明逻辑上已经从条件队列中取消
//这里将其从队列中移除
if (t != null && t.waitStatus != Node.CONDITION)
//该方法就是从条件队列中移除取消的节点
unlinkCancelledWaiters();
t = lastWaiter;
//创建条件队列中的节点,waitStatus为CONDITION(-2)
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
//条件队列为空,设置firstWaiter
firstWaiter = node;
else
//条件队列不为空,将当前节点添加到链表中
t.nextWaiter = node;
//设置lastWaiter为最新入队的节点
lastWaiter = node;
return node;
可以看到条件队列就是一个单向链表,通过nextWaiter指针指向下一个节点,队列的第一个节点就是第一个排队的线程,这个和CLH双向链表和head节点为空Node的结构差别很大。
线程入条件队列之后会释放ReentrantLock的锁,也就是释放state字段,接着判断线程是否在同步队列中,也就是isOnSyncQueue方法:
final boolean isOnSyncQueue(Node node)
//如果ws是CONDITION或者node.prev为null,说明线程在条件队列中
if (node.waitStatus == Node.CONDITION || node.prev == null)
//CLH入队会先设置node的prev,如果为空,说明一定没有入队
return false;
if (node.next != null) // If has successor, it must be on queue
//到这里说明prev!=null,如果next也不为null,结合CHL入队的逻辑
//node一定在CLH队列中,并且是中间节点
return true;
//逻辑到这里的条件是ws!=CONDITION && prev!=null && next==null
//这个条件理论上就代表了node是CLH的尾节点
//但是在节点入CLH队列的时候,是先设置prev,再通过CAS设置tail
//而CAS可能会失败
return findNodeFromTail(node);
首先,如果ws为CONDITION,那么说明线程肯定在条件队列中,否则就要看情况,看什么情况呢?这个需要回顾一下ReentrantLock部分介绍的CLH队列入队操作,这里回顾一下代码片段:
node.prev = t;
if (compareAndSetTail(t, node))
t.next = node;
return t;
入队的逻辑是先设置当前节点的prev节点为队列当前的尾节点,然后通过CAS设置新的尾节点为当前入队节点。那么通过这个逻辑我们可以发现,如果prev!=null&&next!=null,那么node一定在CLH队列中,并且是中间节点;如果prev!=null&&next==null,并不能代表已经是CLH队列中的尾节点,CAS可能会失败,然后继续自旋。所以需要findNodeFromTail方法进一步检查:
private boolean findNodeFromTail(Node node)
Node t = tail;
for (;;)
if (t == node)
return true;
if (t == null)
return false;
t = t.prev;
该方法逻辑很简单,就是从tail节点开始往前寻找,如果找到了对应的节点,那么表示在队列中,否则就不在。因为节点入CLH队列的逻辑是先设置prev,所以prev是可靠的,如果next!=null也可靠,但是next==null则不可靠。
如果判断到节点不在同步队列中,那么还是通过park方法阻塞线程。阻塞的逻辑先看到这里,我们接着来看看唤醒的逻辑,进入nextGeneration方法:
private void nextGeneration()
//唤醒线程
trip.signalAll();
//重置count,开始下一轮分代
count = parties;
generation = new Generation();
到AQS中的ConditionObject中找到signalAll方法:
public final void signalAll()
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignalAll(first);
将条件队列的头结点firstWaiter传入doSignalAll方法:
private void doSignalAll(Node first)
lastWaiter = firstWaiter = null;
do
Node next = first.nextWaiter;
first.nextWaiter = null;
//将条件队列中的节点转入同步队列(CLH)
transferForSignal(first);
first = next;
while (first != null);
该逻辑中将条件队列中的节点按照从头到尾的顺序转入到CLH队列中,核心逻辑在transferForSignal方法中:
final boolean transferForSignal(Node node)
//通过CAS将node.ws修改为0
//如果node.ws不为CONDITION,说明被取消了,直接返回false,不用转入同步队列
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
//自旋CAS入队的逻辑
Node p = enq(node);
int ws = p.waitStatus;
//检查节点状态,如果节点被取消,那么直接唤醒node
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
我们发现节点转入CLH同步队列之后就没有下文了,transferForSignal中的unpark也只是特殊"异常情况"的处理,那么线程是在哪里被唤醒的呢?
不要忘记了整个CyclicBarrier的核心方法dowait的逻辑:先通过ReentrantLock的lock方法加锁,线程进入条件队列后会将state释放,但是此时没有执行释放锁(unlock)的逻辑,条件队列是通过ReentrantLock创建的,ConditionObject是AQS的内部类。当最后一个线程到达(await)之后会将条件队列中的节点"转入"同步队列,最终会到dowait方法的finally中执行lock.unlock(),这个lock.unlock唤醒的就是同步队列中的阻塞线程,按照ReentrantLock的释放逻辑,unlock会唤醒head.next节点,head.next被唤醒后,由于已经换代,那么会从自旋中退出,同样到finally中的unlock逻辑,这样依次循环唤醒同步队列中的所有线程。
总结
在CyclicBarrier初始化的时候,会把parties赋值给count字段,每个线程调用await方法(最终调用dowait方法)的时候,会先通过ReentrantLock上锁,接着会将count减1,如果count被减之后还大于0 ,那么表示还有线程没有就位,就需要将当前线程放入条件等待队列(使用Node构建的一个单向链表),释放lock中的state(只是释放了state,但是没有执行唤醒阻塞线程的逻辑),然后park阻塞;如果count被减之后等于0,那么表示所有线程已经到位,那么最后就位的这个线程会将条件队列中的阻塞线程转移到CLH队列中,然后重置count为parties的值,并且创建一个新的Generation,表示已经换代,如果指定了command,还会同步执行其run方法。接着dowait方法可以返回,但是会在finally块中执行ReentrantLock的unlock方法,会唤醒head.next节点对应的阻塞线程,按照ReentrantLock的逻辑,线程被唤醒之后,该节点会成为新的head节点。一个阻塞线程被唤醒之后会继续执行逻辑,判断到已经换代,那么直接跳出自旋,同样来到finally中的unlock方法,然后唤醒下一个节点对应的阻塞线程,然后该节点又成为新的head节点,接着下一个阻塞线程又被唤醒,就这样一个线程唤醒下一个线程,依次将所有线程唤醒。对此我们可以总结以下几个关键点:
- 一旦一个线程被唤醒之后发现当前分代被打断,那么会抛出BrokenBarrierException异常
- 如果线程在ReentrantLock的lock中被中断过,即使由于lock方法不会抛出异常,但是会自我中断携带中断标识(参考ReentrantLock中的逻辑),在dowait的逻辑中判断到线程被中断过,也会打断当前代,唤醒其它阻塞线程,并且抛出InterruptedException
- 所谓的条件队列和同步队列都是逻辑上的定义,实质上他们都是Node节点,条件队列的头节点为firstWaiter,同步队列(CLH)的头结点为head。换句话说就是一个阻塞线程只有一个Node对象与之对应,它在条件队列中和在同步队列中的node都是同一个对象
- 线程正常的唤醒流程是:最后一个就位线程负责把所有条件队列中的线程添加到同步队列,然后在finally中执行ReentrantLock的unlock方法唤醒同步队列中的head.next,本节点成为新的head节点,然后被唤醒线程同样会到finally的unlock方法中唤醒下一个线程,这样传递唤醒
- 如果一个带超时阻塞的线程被超时唤醒,那么会强制打断当前分代,然后唤醒所有线程,并且抛出TimeoutException异常。这种情况下,其它线程被唤醒后发现是当前分代被打断了,那么会抛出BrokenBarrierException异常
- 如果指定的command在执行过程中出现异常,那么也会打断当前分代,唤醒所有线程,被唤醒线程会抛出BrokenBarrierException异常
上述逻辑是站在一个普遍正常的流程下的描述,事实上一个线程被阻塞之后随时都可能被中断唤醒,被中断唤醒的时候可能在条件队列中,也可能在同步队列中,源码中对于一些特殊情况作出了处理,代码细节点不少,由于篇幅问题本文也就没有每行代码都分析到位~
以上是关于Java并发编程(十六):CyclicBarrier源码分析的主要内容,如果未能解决你的问题,请参考以下文章
Java并发编程(十六):CyclicBarrier源码分析
Java并发编程(十六):CyclicBarrier源码分析