CyclicBarrier

1、CyclicBarrier使用场景：

先来描述一下它的使用场景：有若干个线程，比如说有五个线程，需要它们都到达了某一个点之后才能开始一起执行，也就是说假如其中只有四个线程到达了这个点，还差一个线程没到达，此时这四个线程都会进入等待状态，直到第五个线程也到达了这个点之后，这五个线程才开始一起进行执行状态，是不是这个场景的描述跟CountDownLatch很类似的，下面用一个简单的示例图来感受一下它们两者的区别：

CountDownLatch使用场景图

image.png

CyclicBarrier使用场景图

image.png

所有子线程都已经到达屏障之后，此时屏障就会消失，所有子线程继续执行，若存子线程尚未到达屏障，其他到达了屏障的线程都会进行等待

2、官方文档说明

它是一个同步的工具，能够允许一组线程去互相等待直到都到达了屏障，CyclicBarrier对于涉及到固定大小的线程是非常有用的，线程们必须相互等待。该屏障称之为循环屏障，是因为当等待屏障的线程被释放之后，该屏障能循环使用

3、关于CyclicBarrier的底层执行流程总结：

• 1、初始化CyclicBarrier中的各种成员变量，包括parties、count以及Runnable（可选）；

• 2、当调用await()方法时，底层会先检查计数器是否已经归零，如果是的话，那么就首先执行可选的Runnable，接下来开始下一个generation；（注意：这里只是调用Runnable的run()方法，并不是调用start()方法开启另一个线程）

• 3、在下一个分代中，将会重置count值为parties，并且创建新的Generation实例；

• 4、同时会调用Condition的singalAll方法，唤醒所有在屏障前面等待的线程，让其开始继续执行；（注意：当有可选的Runnable时，是执行完run()方法中的汇总操作，其他线程才会继续执行）

• 5、如果计数器没有归零，那么当前的调用线程将会通过Condition的await方法，在屏障前进行等待；

• 6、以上所有执行流程均在lock锁的控制范围内，不会出现并发情况。

• 7、在下一个分代时，该屏障又可以继续使用，例如计数器是3，线程1，线程2和线程3冲破了当前屏障后，下一个分代的屏障可以去给线程4，线程5和线程6使用，也可以又给线程1，线程2和线程3使用（自己总结的）

4、典型事例

1、当没有可选的Runnable时

当所有线程到达屏障时，不需要进行汇总，最后一个线程到达时，屏障消除，所有线程继续执行

image.png

package com.concurrency2;

import java.util.Random;
import java.util.concurrent.CyclicBarrier;

public class MyTest1 {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
        for(int i = 0;i < 3;i ++) {
            new Thread(() -> {
                try {
                    Thread.sleep((long)(Math.random() * 2000));

                    int randomInt = new Random().nextInt(500);
                    System.out.println("hello " + randomInt);

                    cyclicBarrier.await();

                    System.out.println("world " + randomInt);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

输出

hello 30
hello 471
hello 343
world 343
world 471
world 30

2、当有可选的Runnable时

当所有线程到达屏障时，需要进行汇总操作，等汇总操作进行完，屏障消除，所有线程继续执行

image.png

package com.concurrency2;

import java.util.Random;
import java.util.concurrent.CyclicBarrier;

public class MyTest1 {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
            System.out.println("汇总1 ...");

            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("汇总2 ...");
        });
       //for(int u = 0, u < 2;u ++)//开两次屏障使用
        for(int i = 0;i < 3;i ++) {
            new Thread(() -> {
                try {
                    Thread.sleep((long)(Math.random() * 2000));

                    int randomInt = new Random().nextInt(500);
                    System.out.println("hello " + randomInt);

                    cyclicBarrier.await();

                    System.out.println("world " + randomInt);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

输出

hello 4
hello 229
hello 73
汇总1 ...
汇总2 ...
world 73
world 229
world 4

5、CyclicBarrier源代码分析

前面讲过CountDownLatch是基于AQS实现的；而CyclicBarrier是基于ReentrantLock重入锁实现的，当然ReentrantLock也是基于AQS实现的，非要说CyclicBarrier也是基于AQS实现的也不为过。

1、重要成员变量

    / /可以理解为初始化时 需要阻塞的任务个数
    private final int parties;
    / /剩余需要等待的任务个数，初始值为parties，直到为0时依次唤醒所有被阻塞的任务线程。
    private int count;
 
    / /每次对“栅栏”的主要成员变量进行变更操作，都应该加锁
    private final ReentrantLock lock = new ReentrantLock();
    / /用于阻塞和唤醒任务线程
   private final Condition trip = lock.newCondition();
 
    / /在所有线程被唤醒前，需要执行的一个Runable对应的run方法
    private final Runnable barrierCommand;
    / /用于表示“栅栏”当前的状态
    private Generation generation = new Generation();

2、构造方法

CyclicBarrier有两个重载的构造方法，一个是不带Runnable参数，另一个带有Runnable参数。本质上都会调用带Runnable参数的构造方法进行实例化，这里只贴出带Runnable参数的构造方法实现：

public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;    / /为了实现复用，进行备份
        this.count = parties;   / /初始化，待阻塞的任务总数
        this.barrierCommand = barrierAction;   / /初始化
    }

3、核心方法

await()方法有两层含义：

1、先检查前面是否已经有count个线程了，如果没有线程则会进入等待状态
2、当检测到屏障已经有count个线程了，则所有线程会冲出屏障继续执行（如果有Runnable参数的构造方法先执行汇总方法）

int index = --count;操作很明显不是原子性的，如果在多线程中不加lock肯定会出问题

private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation();
    }

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(); //加锁
        try {
            final Generation g = generation;
 
            if (g.broken)
                throw new BrokenBarrierException();
            //有一个线程线程被中断，整个CyclicBarrier将不可用
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }
 
            int index = --count; //待等待的任务数减1
            if (index == 0) {  // 如果待等待的任务数减至0，依次唤醒所有线程
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();//唤醒前先执行Runnable对象的run方法
                    ranAction = true;
                    nextGeneration();//重置整个CyclicBarrier，方便下次重用
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
 
            //如果待等待的任务数大于0，进行线程阻塞，直到count为0时被唤醒
            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)//正常被唤醒，generation会被新建
                    return index;
 
                if (timed && nanos <= 0L) {//延迟阻塞时间到后唤醒
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
}