Java并发工具类 - CyclicBarrier

作者: 怡红快绿 | 来源:发表于2019-03-07 17:07 被阅读0次

并发编程艺术-8
Java并发工具类 - CyclicBarrier
JAVA多线程-CyclicBarrier同步屏障
高并发（9）- 线程并发工具类-CyclicBarrier
Java并发工具包——CyclicBarrier
CountDownLatch和CyclicBarrier
Java并发工具类（栅栏CyclicBarrier）
Java并发工具类--CyclicBarrier详解
第八章并发工具类
并发工具类：CyclicBarrier

一、CyclicBarrier是什么

CyclicBarrier指的就是循环屏障，虽然这个叫法很奇怪，但是确能很好地表示它的作用。

其作用在JDK注释中是这样描述的：

A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.
CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other.
The barrier is called cyclic because it can be re-used after the waiting threads are released.

CyclicBarrier是一个同步辅助类，它允许一组线程相互等待直到所有线程都到达一个公共的屏障点。
在程序中有固定数量的线程，这些线程有时候必须等待彼此，这种情况下，使用CyclicBarrier很有帮助。
这个屏障之所以用循环修饰，是因为在所有的线程释放彼此之后，这个屏障是可以重新使用的。

CyclicBarrier的简单理解

其实，我更喜欢[人满发车]这个词来理解CyclicBarrier的作用：

长途汽车站提供长途客运服务。
当等待坐车的乘客到达20人时，汽车站就会发出一辆长途汽车，让这20个乘客上车走人。
等到下次等待的乘客又到达20人是，汽车站就会又发出一辆长途汽车。

二、主要方法

CyclicBarrier(parties)
初始化相互等待的线程数量的构造方法。
CyclicBarrier(parties,Runnable barrierAction)
初始化相互等待的线程数量以及屏障线程的构造方法。
屏障线程的运行时机：等待的线程数量=parties之后，CyclicBarrier打开屏障之前。
举例：在分组计算中，每个线程负责一部分计算，最终这些线程计算结束之后，交由屏障线程进行汇总计算。
getParties()
获取CyclicBarrier打开屏障的线程数量，也成为方数。
getNumberWaiting()
获取正在CyclicBarrier上等待的线程数量。
await()
在CyclicBarrier上进行阻塞等待，直到发生以下情形之一：
在CyclicBarrier上等待的线程数量达到parties，则所有线程被释放，继续执行。
当前线程被中断，则抛出InterruptedException异常，并停止等待，继续执行。
其他等待的线程被中断，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。
其他等待的线程超时，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。
其他线程调用CyclicBarrier.reset()方法，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。
await(timeout,TimeUnit)
在CyclicBarrier上进行限时的阻塞等待，直到发生以下情形之一：
在CyclicBarrier上等待的线程数量达到parties，则所有线程被释放，继续执行。
当前线程被中断，则抛出InterruptedException异常，并停止等待，继续执行。
当前线程等待超时，则抛出TimeoutException异常，并停止等待，继续执行。
其他等待的线程被中断，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。
其他等待的线程超时，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。
其他线程调用CyclicBarrier.reset()方法，则当前线程抛出BrokenBarrierException异常，并停止等待，继续执行。
isBroken()
获取是否破损标志位broken的值，此值有以下几种情况：
CyclicBarrier初始化时，broken=false，表示屏障未破损。
如果正在等待的线程被中断，则broken=true，表示屏障破损。
如果正在等待的线程超时，则broken=true，表示屏障破损。
如果有线程调用CyclicBarrier.reset()方法，则broken=false，表示屏障回到未破损状态。
reset()
使得CyclicBarrier回归初始状态，直观来看它做了两件事：
如果有正在等待的线程，则会抛出BrokenBarrierException异常，且这些线程停止等待，继续执行。
将是否破损标志位broken置为false。

三、模拟多线程分组计算

实现方式：

用3个线程分别计算：定义一个大小为3的线程池。
计算结果进行合并：定义一个屏障线程，将上面3个线程计算的子结果信息合并。

代码：

首先创建一个屏障线程，用于汇总多个子结果

final CyclicBarrier barrier2 = new CyclicBarrier(3, new Runnable() {
    @Override
    public void run() {
        //汇总结果
        System.out.println("开始汇总计算结果");
    }
});

模拟3个子线程执行计算，并得到3个子结果

final ExecutorService executorService2 = Executors.newFixedThreadPool(3);
for (int i = 0; i < 3; i++) {
    final int finalI = i;
    executorService2.submit(
            new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("线程 "+ finalI +" 开始计算…计算完成");
                        barrier2.await();
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            });
}

以下是完整代码

public static void main(String... args) {
    final CyclicBarrier barrier2 = new CyclicBarrier(3, new Runnable() {
        @Override
        public void run() {
            //汇总结果
            System.out.println("开始汇总计算结果……汇总完成");
        }
    });
    final ExecutorService executorService2 = Executors.newFixedThreadPool(3);
    for (int i = 0; i < 3; i++) {
        final int finalI = i;
        executorService2.submit(
                new Runnable() {
                    @Override
                    public void run() {
                        try {
                            System.out.println("线程 " + finalI + " 开始计算…计算完成");
                            barrier2.await();
                        } catch (InterruptedException | BrokenBarrierException e) {
                            e.printStackTrace();
                        }
                    }
                });
    }
  //最后记得要关闭线程池
    executorService2.shutdown();
}