JUC回顾之-CyclicBarrier底层实现和原理

Posted 2020-08-19 积淀

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JUC回顾之-CyclicBarrier底层实现和原理相关的知识，希望对你有一定的参考价值。

1.CyclicBarrier 字面意思是可循环（Cyclic）使用的屏障（Barrier）。它要做的事情是让一组线程到达一个屏障（同步点）时被阻塞，直到最后一个线程到达屏障时候，屏障才会开门。所有被屏障拦截的线程才会运行。

2.常用的方法：

 
CyclicBarrier(int parties)
创建一个新的 CyclicBarrier，它将在给定数量的参与者（线程）处于等待状态时启动，但它不会在启动 barrier 时执行预定义的操作。

CyclicBarrier(int parties, Runnable barrierAction)
创建一个新的 CyclicBarrier，它将在给定数量的参与者（线程）处于等待状态时启动，并在启动 barrier 时执行给定的屏障操作，该操作由最后一个进入 barrier 的线程执行。

int await()
在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。

int await(long timeout, TimeUnit unit)
在所有参与者都已经在此屏障上调用 await 方法之前将一直等待,或者超出了指定的等待时间。

int getNumberWaiting()
返回当前在屏障处等待的参与者数目。

int getParties()
返回要求启动此 barrier 的参与者数目。

boolean isBroken()
查询此屏障是否处于损坏状态。

void reset() 
将屏障重置为其初始状态。如果调用了该函数，则在等待的线程将会抛出BrokenBarrierException异常。

3.底层原理实现

CyclicBarrier是由ReentrantLock可重入锁和Condition共同实现的。

具体实现源码如下：

//1.CyclicBarrier构造方法
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    // parties表示“必须同时到达barrier的线程个数”。
    this.parties = parties;
    // count表示“处在等待状态的线程个数”。
    this.count = parties;
    // barrierCommand表示“parties个线程到达barrier时，会执行的动作”。
    this.barrierCommand = barrierAction;
}

1. await源码分析：

 

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 {
    final ReentrantLock lock = this.lock;
    // 获取“独占锁(lock)”
    lock.lock();
    try {
        // 保存“当前的generation”
        final Generation g = generation;

        // 若“当前generation已损坏”，则抛出异常。
        if (g.broken)
            throw new BrokenBarrierException();

        // 如果当前线程被中断，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程。
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }

       // 将“count计数器”-1
       int index = --count;
       // 如果index=0，则意味着“有parties个线程到达barrier”。
       if (index == 0) {  // tripped
           boolean ranAction = false;
           try {
               // 如果barrierCommand不为null，则执行该动作。
               final Runnable command = barrierCommand;
               if (command != null)
                   command.run();
               ranAction = true;
               // 唤醒所有等待线程，并更新generation。
               nextGeneration();
               return 0;
           } finally {
               if (!ranAction)
                   breakBarrier();
           }
       }

        // 当前线程一直阻塞，直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，
        // 当前线程才继续执行。
        for (;;) {
            try {
                // 如果不是“超时等待”，则调用await()进行等待；否则，调用awaitNanos()进行等待。
                if (!timed)
                    trip.await();
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                // 如果等待过程中，线程被中断，则执行下面的函数。
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    Thread.currentThread().interrupt();
                }
            }

            // 如果“当前generation已经损坏”，则抛出异常。
            if (g.broken)
                throw new BrokenBarrierException();

            // 如果“generation已经换代”，则返回index。
            if (g != generation)
                return index;

            // 如果是“超时等待”，并且时间已到，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程，并抛出TimeoutException异常
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        // 释放“独占锁(lock)”
        lock.unlock();
    }
}

实例如下：

1.await()方法使用

public class CyclicBarrierTest {

    private static int SIZE = 5;

    private static CyclicBarrier cb;

    public static void main(String[] args) {
        cb = new CyclicBarrier(SIZE);
        for (int i = 0; i < SIZE; i++) {
            new MyTask().start();
        }

    }

    static class MyTask extends Thread {
        @Override
        public void run() {
            try {

                System.out.println("线程" + Thread.currentThread().getName() + "正在执行同一个任务");
                // 以睡眠来模拟几个线程执行一个任务的时间
                Thread.sleep(1000);
                System.out.println("线程" + Thread.currentThread().getName() + "执行任务完成，等待其他线程执行完毕");
                // 用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；
                cb.await();

            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println("所有线程写入完毕");

        }

    }

}

结果

线程Thread-0正在执行同一个任务
线程Thread-2正在执行同一个任务
线程Thread-1正在执行同一个任务
线程Thread-3正在执行同一个任务
线程Thread-4正在执行同一个任务
线程Thread-3执行任务完成，等待其他线程执行完毕
线程Thread-4执行任务完成，等待其他线程执行完毕
线程Thread-0执行任务完成，等待其他线程执行完毕
线程Thread-1执行任务完成，等待其他线程执行完毕
线程Thread-2执行任务完成，等待其他线程执行完毕
所有线程写入完毕
所有线程写入完毕
所有线程写入完毕
所有线程写入完毕
所有线程写入完毕

2.使用 barrier.await(2000, TimeUnit.MILLISECONDS)方法：

public class CyclicBarrierWriteDataWaitTimeOutTest {

    private static final int THREAD_NUM = 5;

    private static final Random random = new Random();

    public static void main(String[] args) throws InterruptedException {

        // 使用构造方法：public CyclicBarrier(int parties, Runnable barrierAction)
        // 参数parties表示一共有多少线程参与这次“活动”,barrierAction是可选的，用来指定当所有线程都完成这些必须的“任务”之后要干的其他事情
        CyclicBarrier barrier = new CyclicBarrier(THREAD_NUM, new Runnable() {

            @Override
            public void run() {
                // 最后写完数据的线程，会执行这个任务
                System.out.println(Thread.currentThread().getId() + ":所有线程写数据完毕！");
            }
        });

        // 启动5个线程，写数据
        for (int i = 0; i < THREAD_NUM; i++) {
            if (i < THREAD_NUM - 1) {
                Thread t = new Thread(new MyTask(barrier));
                t.start();
            } else {
                // 最后一个线程延迟3秒执行
                Thread.sleep(3000);
                Thread t = new Thread(new MyTask(barrier));
                t.start();
            }

        }

    }

    /**
     * 
     * (线程类)
     *
     * <p>
     * 修改历史:                                            <br>  
     * 修改日期            修改人员       版本             修改内容<br>  
     * -------------------------------------------------<br>  
     * 2016年8月26日 上午11:21:39   user     1.0        初始化创建<br>
     * </p> 
     *
     * @author        Peng.Li 
     * @version        1.0  
     * @since        JDK1.7
     */
    static class MyTask extends Thread {

        private CyclicBarrier barrier;

        public MyTask(CyclicBarrier barrier) {
            this.barrier = barrier;
        }

        @Override
        public void run() {

            int time = random.nextInt(1000);
            System.out.println(Thread.currentThread().getId() + "：需要" + time + "毫秒的时间写入数据");

            try {
                Thread.sleep(time);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            System.out.println(Thread.currentThread().getId() + "：写入数据完毕，等待其他线程写入数据");
            try {
                // 用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；
                // 等待所有线程都调用过此函数才能继续后续动作
                // 只等待2s，那么最后一个线程3秒才执行，则必定会超时
                barrier.await(2000, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getId() + "：所有线程写入数据完毕，继续处理其他任务...");

        }

    }

}

从结果分析:

可以看到，前面四个线程等待最后一个线程超时了，这个时候这四个线程不会再等待最后一个线程写入完毕，而是直接抛出BrokenBarrierException

异常，继续执行后续的动作。最后一个线程完成写入数据操作后也继续了后续的动作。

需要理解的是：最后一个线程发生超时的异常，其他的线程不会继续等待，而是去执行其他的任务。

Thread-1：需要650毫秒的时间写入数据
Thread-3：需要297毫秒的时间写入数据
Thread-5：需要755毫秒的时间写入数据
Thread-7：需要79毫秒的时间写入数据
Thread-7：写入数据完毕，等待其他线程写入数据
Thread-3：写入数据完毕，等待其他线程写入数据
Thread-1：写入数据完毕，等待其他线程写入数据
Thread-5：写入数据完毕，等待其他线程写入数据
Thread-7：所有线程写入数据完毕，继续处理其他任务...
Thread-1：所有线程写入数据完毕，继续处理其他任务...
java.util.concurrent.TimeoutException
Thread-3：所有线程写入数据完毕，继续处理其他任务...
Thread-5：所有线程写入数据完毕，继续处理其他任务...
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:250)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
    at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
    at java.lang.Thread.run(Thread.java:745)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:243)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
    at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
    at java.lang.Thread.run(Thread.java:745)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:243)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
    at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
    at java.lang.Thread.run(Thread.java:745)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:243)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
    at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
    at java.lang.Thread.run(Thread.java:745)
Thread-9：需要164毫秒的时间写入数据
Thread-9：写入数据完毕，等待其他线程写入数据java.util.concurrent.BrokenBarrierException
Thread-9：所有线程写入数据完毕，继续处理其他任务...

    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:200)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:427)
    at concurrentMy.CountAndCyclicAndSemaphore.cyclibarrier.test.CyclicBarrierWriteDataWaitTimeOutTest$MyTask.run(CyclicBarrierWriteDataWaitTimeOutTest.java:299)
    at java.lang.Thread.run(Thread.java:745)

以上是关于JUC回顾之-CyclicBarrier底层实现和原理的主要内容，如果未能解决你的问题，请参考以下文章

JUC回顾之-ScheduledThreadPoolExecutor底层实现原理和应用

Java多线程系列--“JUC锁”10之 CyclicBarrier原理和示例

Juc21_强大的三个工具类CountDownLatch 闭锁 CyclicBarrier Semaphore

JUC并发编程共享模式之工具 JUC CyclicBarrier（循环栅栏与CountdownLatch最大的不同是可以重值倒计时） -- CyclicBarrier介绍使用注意事项

JUC - 多线程之同步辅助CountDownLatch，CyclicBarrier，Semaphore

java中的JUC组件（SemaphoreCountDownLatchCyclicBarrier）