CyclicBarrier

    CyclicBarrier翻译过来是“可循环利用的屏障“，CyclicBarrier 作用是让一组线程相互等待，当达到一个共同点时，所有之前等待的线程再继续执行，且 CyclicBarrier 功能可重复使用。

image.png

代码示例

public class CyclicBarrierDemo {


    public static void main(String[] args) {
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(20);
        int corePoolSize = 10;
        int maximumPoolSize = 10;
        long keepAliveTime = 60L;

        CyclicBarrier cyclicBarrier = new CyclicBarrier(10, ()-> {
            System.out.println(Thread.currentThread().getName()+"已经到达检查点");
        });

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

        for(int i=0; i<10; i++) {
            threadPoolExecutor.submit(() -> {
                try {
                    System.out.println(Thread.currentThread().getName()+"已经到达屏障");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName()+"已经冲破屏障");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }
        threadPoolExecutor.shutdown();

    }
}

执行结果

pool-1-thread-2已经到达屏障
pool-1-thread-4已经到达屏障
pool-1-thread-3已经到达屏障
pool-1-thread-6已经到达屏障
pool-1-thread-1已经到达屏障
pool-1-thread-5已经到达屏障
pool-1-thread-7已经到达屏障
pool-1-thread-8已经到达屏障
pool-1-thread-9已经到达屏障
pool-1-thread-10已经到达屏障
pool-1-thread-10已经到达检查点
pool-1-thread-10已经冲破屏障
pool-1-thread-2已经冲破屏障
pool-1-thread-3已经冲破屏障
pool-1-thread-4已经冲破屏障
pool-1-thread-9已经冲破屏障
pool-1-thread-8已经冲破屏障
pool-1-thread-7已经冲破屏障
pool-1-thread-5已经冲破屏障
pool-1-thread-1已经冲破屏障
pool-1-thread-6已经冲破屏障

主要结构

public class CyclicBarrier {

    // CyclicBarrier是可复用的，Generation用于标记更新换代
    private static class Generation {
        boolean broken = false;
    }

    /** The lock for guarding barrier entry */
    private final ReentrantLock lock = new ReentrantLock();
    /** Condition to wait on until tripped */
    private final Condition trip = lock.newCondition();
    /** The number of parties */
    private final int parties;
    /* The command to run when tripped */
    private final Runnable barrierCommand;
    /** The current generation */
    private Generation generation = new Generation();

    /**
     * Number of parties still waiting. Counts down from parties to 0
     * on each generation.  It is reset to parties on each new
     * generation or when broken.
     */
     // 可以简单理解为还未到达屏障的成员数
    private int count;
}

构造函数

// 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
    返回值：当前线程的到达索引，0表示最后一个到达

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();
            // 线程被中断，调用breakBarrier打破屏障
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

            int index = --count;
            // 说明该线程是最后一个到达的线程
            if (index == 0) {  // tripped
                // 标志任务是否执行成功
                boolean ranAction = false;
                try {
                   
                    final Runnable command = barrierCommand;
                    // 任务不是空的话执行任务
                    if (command != null)
                        command.run();
                    ranAction = true;
                    // 所有线程都到达屏障了，开启下一代
                    nextGeneration();
                    return 0;
                } finally {
                    // 任务执行失败，打破屏障
                    if (!ranAction)
                        breakBarrier();
                }
            }

            // loop until tripped, broken, interrupted, or timed out
            for (;;) {
                try {
                    // 没有设置超时时间，当前线程等待
                    if (!timed)
                        trip.await();
                  // 设置了超时时间，当前线程等到nanos时间
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    // 线程阻塞期间被中断，朝代没有换，屏障没有被打破，则breakBarrier打破屏障
                    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();
                    }
                }
                
                // 线程被唤醒，如果屏障被打断则抛出BrokenBarrierException
                if (g.broken)
                    throw new BrokenBarrierException();
              
               // 已经更新换代过了，说明最后一个线程也到达了
               // 那么直接返回当前线程的到达屏障时的索引就行
                if (g != generation)
                    return index;
                // 如果设置了超时时间，并且超时时间小于0，则打破屏障，并抛出TimeoutException
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

    屏障被打破：是指调用breakBarrier 方法把broken被置为true，线程被中断、barrierCommand任务执行失败、超时、调用reset方法都会打破屏障

复位屏障

    将屏障重置为其初始状态。如果任何一方当前正在等待障碍，他们将抛出BrokenBarrierException

 public void reset() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 在当前代打破屏障
            breakBarrier();   // break the current generation
            // 开启新代
            nextGeneration(); // start a new generation
        } finally {
            lock.unlock();
        }
    }

开启新代

  private void nextGeneration() {
        // 唤醒在trip条件等待队列中的所有线程
        trip.signalAll();
        // set up next generation
        // 初始化等待的成员数
        count = parties;
        // new一个新Generation
        generation = new Generation();
    }

打破屏障

private void breakBarrier() {
        // 设置broken标志位
        generation.broken = true;
        count = parties;
        // 唤醒在trip条件等待队列中的所有线程
        trip.signalAll();
    }