java源码 - CyclicBarrier

开篇

• CyclicBarrier是一个同步工具类，它允许一组线程互相等待，直到到达某个公共屏障点。与CountDownLatch不同的是该barrier在释放等待线程后可以重用，所以称它为循环（Cyclic）的屏障（Barrier）。

• CyclicBarrier支持一个可选的Runnable命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作很有用。

• CyclicBarrier的内部实现逻辑基于ReentrantLock实现，可以理解为ReentrantLock的上层应用者，通过ReentrantLock的Condtion实现线程的休眠和唤醒。

CyclicBarrier用法demo

public class Test {
    public static void main(String[] args) {
        int N = 4;
        CyclicBarrier barrier  = new CyclicBarrier(N);
        for(int i=0;i<N;i++)
            new Writer(barrier).start();
    }
    static class Writer extends Thread{
        private CyclicBarrier cyclicBarrier;
        public Writer(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }
 
        @Override
        public void run() {
            System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");
            try {
                Thread.sleep(5000);      //以睡眠来模拟写入数据操作
                System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }catch(BrokenBarrierException e){
                e.printStackTrace();
            }
            System.out.println("所有线程写入完毕，继续处理其他任务...");
        }
    }
}

线程Thread-0正在写入数据...
线程Thread-3正在写入数据...
线程Thread-2正在写入数据...
线程Thread-1正在写入数据...
线程Thread-2写入数据完毕，等待其他线程写入完毕
线程Thread-0写入数据完毕，等待其他线程写入完毕
线程Thread-3写入数据完毕，等待其他线程写入完毕
线程Thread-1写入数据完毕，等待其他线程写入完毕
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...

CyclicBarrier类定义

• parties记录一共等待执行个数，count记录依然等待执行的个数。
• barrierCommand记录所有待执行的完成后由最后一个线程执行的完成的命令。
• Generation的代的概念来实现CyclicBarrier的复用。
• 构造函数负责初始化parties、count、barrierCommand的核心变量。

public class CyclicBarrier {
    // 代的类定义
    private static class Generation {
        boolean broken = false;
    }

    // 内部通过ReentrantLock实现线程安全的等待
    private final ReentrantLock lock = new ReentrantLock();
    // 内部通过Lock的condition实现所有waiter的信号通知
    private final Condition trip = lock.newCondition();
    // 所有等待执行的个数
    private final int parties;
    // 所有等待线程都完成任务后由最后一个线程执行的命令
    private final Runnable barrierCommand;
   
    // 通过代的概念实现复用
    private Generation generation = new Generation();

    // 还在等待的个数
    private int count;

    // 核心构造函数
    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);
    }

CyclicBarrier工作原理

• CyclicBarrier通过ReentrantLock来保证线程休眠和唤醒的通信。
• 在执行过程中会对等待计数进行减一操作，值不为0当前线程进入休眠等待其他线程唤醒
• 在执行过程中会对等待计数进行减一操作，值为0当前线程直接执行barrierCommand并且通过nextGeneration方法唤醒其他等待线程
• 线程的休眠和唤醒都是基于ReentrantLock来实现的

    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }


    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        // 通过lock来保证线程安全
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Generation g = generation;
            // 判断generation过期的情况
            if (g.broken)
                throw new BrokenBarrierException();
            // 判断线程中断情况
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

            // 递减待执行的个数计数
            int index = --count;
            // 所有待执行任务完成后执行barrierCommand命令
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    // barrierCommand命令不为null的时候执行该命令
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    // 已经执行了barrierCommand
                    ranAction = true;
                    // 重置generation用以复用并且唤醒所有等待的线程
                    //       private void nextGeneration() {
                    //           trip.signalAll();
                    //           count = parties;
                    //           generation = new Generation();
                    //        }
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }

            // 如果count的值不为0，那么当前线程就开始进入等待
            // 外层通过lock占用锁，内层通过wait()进入休眠并释放锁
            for (;;) {
                try {
                    if (!timed)
                        // private final Condition trip = lock.newCondition();
                        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();
        }
    }

参考文章

Java并发编程：CountDownLatch、CyclicBarrier和Semaphore