Java 并发编程之 CyclicBarrier

作用

允许一组线程互相等待，以达到共同的障碍点，再执行CyclicBarrier中指定Runable任务。

人满开车：到3（设置人数）就开车。

构造函数

parties: 线程必须调用await的次数。

barrierAction：线程都到达障碍点后执行的操作。（无此参数的构造表示：达到障碍点后不做任何操作。）

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

await() 方法

如果当前线程不是最后一个到达障碍点的，出于调度目的而将此线程禁用，并处于休眠状态，直到以下事件发生：

• 最后一个线程到达障碍点；
• 其他线程打断了当前线程；
• 其他线程打断了处于等待的线程；
• 其他线程等待障碍点时超时；
• 重置了障碍点

barrierAction

该Runnable任务执行在最后一个到达障碍点的线程中。并且只有在执行完该任务后其余等待线程才继续后续操作。

案例

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+" 开始CyclicBarrier");
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3,() -> {
            System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", 完成所有接口请求, 合并数据---开始");
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", 完成所有接口请求, 合并数据---结束");
        });
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", 模拟请求API-1, 耗时2秒, 完成, 等待其他接口返回");
                cyclicBarrier.await();
                System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", CyclicBarrier释放了, API-1");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(4);
                System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", 模拟请求API-2, 耗时4秒, 完成, 等待其他接口返回");
                cyclicBarrier.await();
                System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", CyclicBarrier释放了, API-2");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", 模拟请求API-3, 耗时3秒, 完成, 等待其他接口返回");
                cyclicBarrier.await();
                System.out.println(LocalDateTime.now() + ": ThreadId: " + Thread.currentThread().getId()+", CyclicBarrier释放了, API-3");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }
}

输出：

2020-01-17T14:43:26.314: ThreadId: 1 开始CyclicBarrier
2020-01-17T14:43:28.325: ThreadId: 12, 模拟请求API-1, 耗时2秒, 完成, 等待其他接口返回
2020-01-17T14:43:29.325: ThreadId: 14, 模拟请求API-3, 耗时3秒, 完成, 等待其他接口返回
2020-01-17T14:43:30.325: ThreadId: 13, 模拟请求API-2, 耗时4秒, 完成, 等待其他接口返回
2020-01-17T14:43:30.325: ThreadId: 13, 完成所有接口请求, 合并数据---开始
2020-01-17T14:43:33.326: ThreadId: 13, 完成所有接口请求, 合并数据---结束
2020-01-17T14:43:33.326: ThreadId: 13, CyclicBarrier释放了, API-2
2020-01-17T14:43:33.326: ThreadId: 12, CyclicBarrier释放了, API-1
2020-01-17T14:43:33.326: ThreadId: 14, CyclicBarrier释放了, API-3