Java并发编程(十六)：CyclicBarrier源码分析

Posted 2023-03-01 黄智霖-blog

前言

  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异常

  上述逻辑是站在一个普遍正常的流程下的描述，事实上一个线程被阻塞之后随时都可能被中断唤醒，被中断唤醒的时候可能在条件队列中，也可能在同步队列中，源码中对于一些特殊情况作出了处理，代码细节点不少，由于篇幅问题本文也就没有每行代码都分析到位~