JUC CyclicBarrier 分析

Myth丶恋晨 2022-05-21 23:29 162阅读 0赞

## 基本介绍 ##

`CyclicBarrier`可实现多个线程同时等待，达到一个共同临界点，才一起往下执行，并且，可以在达到共同临界点的时候，触发一个action。这个同步组件实现的功能看似与`CountDownLatch`一样，但是与`CountDownLatch`只能被使用一次不一样，`CyclicBarrier`可被重复多次使用。

像`CountDownLatch`、`Semaphore`、`Lock`等同步组件，都是通过直接扩展AQS来完成实现，但是`CyclicBarrier`不同，它间接地依赖了`ReentrantLock`以及等待通知机制实现`Condition`来实现它的功能。

> 在后面源码分析里面会涉及到它的实现原理

示例:

package com.crazypig.juc;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class CyclicBarrierTest { 
    
        public static void main(String[] args) throws InterruptedException {
    
            final int count = 3;
    
            Runnable barrierAction = new Runnable() {
    
                @Override
                public void run() {
                    System.out.println("run barrier action!");
                }
            };
    
            final CyclicBarrier barrier = new CyclicBarrier(count, barrierAction);
    
            Runnable r = new Runnable() {
    
                @Override
                public void run() {
                    try {
                        System.out.println(Thread.currentThread().getName() + " wait for barrier...");
                        barrier.await();
                        System.out.println(Thread.currentThread().getName() + " end!");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            };
    
            Thread[] thds = new Thread[count];
            for (int i = 0; i < thds.length; i++) {
                thds[i] = new Thread(r, "thd" + i);
                thds[i].start();
            }
    
            for (int i = 0; i < thds.length; i++) {
                thds[i].join();
            }
    
        }
    
    }

## 源码分析 ##

前面提到，`CyclicBarrier`内部利用了`ReentrantLock`和`Condition`来实现其功能。

其功能实现的关键点分析总结如下：

*  内部定义一个计数`parties`，使用lock控制parties的并发操作，当每一次（通常一个线程调用一次）调用await方法，都使得parties - 1，若parties不为0，则线程阻塞等待，阻塞等待通过Condition的await来实现
 *  最后一个线程调用await会使得parties变为0，此时，最后一个线程会负责调用barrierAction（前面提到的，当多个线程达到一个共同临界点【parties= 0的时候】，会触发一个action），然后再调用Condition的signalAll来唤醒前面阻塞等待的其他线程
 *  通过内部定义的`Generation`和重置parties来达到重复使用的功能（重置parties的动作在最后一个await线程唤醒其他线程之后进行）

下面直接透过源码来看其实现原理，代码不多，直接通过在源码上+注释的方式来分析：

public class CyclicBarrier { 
    
        private static class Generation { 
            boolean broken = false;
        }
    
        /** 维护parties的并发-1操作，同时实现阻塞等待(Condition) */
        private final ReentrantLock lock = new ReentrantLock();
        /** await时的条件等待,最后一个await线程负责唤醒其他阻塞等待的线程(signalAll) */
        private final Condition trip = lock.newCondition();
        /** 定义所有await线程个数 */
        private final int parties;
        /* 定义共同达到临界点signalAll之前，首先执行的动作 */
        private final Runnable barrierCommand;
        /** 每一次循环使用CyclicBarrier都会new一个新的Generation实例，该实例还用于控制异常处理 */
        private Generation generation = new Generation();
    
        private int count;
    
        // 由最后一个await线程调用，唤醒所有await的线程，并重置状态，生成新的Generation实例
        private void nextGeneration() {
            // signal completion of last generation
            trip.signalAll();
            // set up next generation
            count = parties;
            generation = new Generation();
        }
    
        // 异常处理，唤醒所有await线程，而后响应BrokenBarrierException
        private void breakBarrier() {
            generation.broken = true;
            count = parties;
            trip.signalAll();
        }
    
        // 由await调用的主逻辑
        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();
                }
    
               int index = --count;
               if (index == 0) {  // 最后一个线程调用await执行的逻辑分支
                   boolean ranAction = false;
                   try {
                        // 在当前线程（最后一个线程）执行barrierCommand(if exists)
                       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
                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;
    
                    // 超时，同样唤醒所有其他等待的线程，使其后续感知BrokenBarrierException
                    if (timed && nanos <= 0L) {
                        breakBarrier();
                        throw new TimeoutException(); // 同时该线程抛TimeoutException给上层感知
                    }
                }
            } finally {
                lock.unlock();
            }
        }
    
        public CyclicBarrier(int parties, Runnable barrierAction) {
            if (parties <= 0) throw new IllegalArgumentException();
            this.parties = parties;
            this.count = parties;
            this.barrierCommand = barrierAction;
        }
    
        public CyclicBarrier(int parties) {
            this(parties, null);
        }
    
        public int getParties() {
            return parties;
        }
    
        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));
        }
    
        public boolean isBroken() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return generation.broken;
            } finally {
                lock.unlock();
            }
        }
    
        public void reset() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                breakBarrier();   // break the current generation
                nextGeneration(); // start a new generation
            } finally {
                lock.unlock();
            }
        }
    
        // 获取当前阻塞等待在await上的线程
        public int getNumberWaiting() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return parties - count;
            } finally {
                lock.unlock();
            }
        }
    }

## 关于BrokenBarrierException ##

以下几种情况会使得等待线程抛BrokenBarrierException:

*  其他await线程被interrupt
 *  其他线程调用了await(timeout, timeunit)并发生了超时
 *  其他线程调用了reset方法