CyclicBarrier 栅栏 原理，应用场景

Posted 2022-12-23 liumy

栅栏类似于闭锁，它能阻塞一组线程直到某个事件发生。

栅栏与闭锁的关键区别 CyclicBarrier和CountDownLatch的区别

在于，所有线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。

我自己写了几个例子，加上自己的理解总结出几个不同。

1. CyclicBarrier 方法多，可以用reset()方法来重置CyclicBarrier，让栅栏可以反复用。而CountDownLatch如果count变为0了，那么只能保持在这个0的最终状态，不能重新再用。

2. CyclicBarrier 是让一组线程等待某个事件发生，如果发生了，这组线程可以继续执行；CountDownLatch是一个线程或多个线程等待一组线程执行完毕。不同的点就在于当count变为0之后，CyclicBarrier是让这组线程不再阻塞，而继续执行；而CountDownLatch是让等待的线程不阻塞，继续执行。

 下面是CyclicBarrier例子。

有三个类。下面是一个开会的例子。首先是组员到达会议室。等到所有组员都到了之后，领导才开始开会。

MeetingLeaderTask:  领导线程。

OpenMeetingTask, 组员线程。

TestOpenMeeting 测试线程

package com.citi.test.mutiplethread.demo5;

public class MeetingLeaderTask implements Runnable 
    @Override
    public void run() 
        System.out.println("**********领导开始开会***********");
        try 
            Thread.sleep(2000);
         catch (InterruptedException e) 
            // TODO Auto-generated catch block
            e.printStackTrace();
        
    

View Code

package com.citi.test.mutiplethread.demo5;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class OpenMeetingTask implements Runnable 
    private final CyclicBarrier barrier;
    private final String name;
    private final int arriveTime;

    public OpenMeetingTask(CyclicBarrier barrier,String name,int arriveTime) 
        this.barrier=barrier;
        this.name=name;
        this.arriveTime=arriveTime;
    

    @Override
    public void run() 
        try 
            Thread.sleep(arriveTime*1000);
            System.out.println(name+"到达会议室");
            barrier.await();
            System.out.println(name+"开始开会。。。");
         catch (InterruptedException e) 
            // TODO Auto-generated catch block
            e.printStackTrace();
         catch (BrokenBarrierException e) 
            // TODO Auto-generated catch block
            e.printStackTrace();
        
    

View Code

package com.citi.test.mutiplethread.demo5;

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestOpenMeeting 
    public static void main(String[] args) 
        CyclicBarrier barrier=new CyclicBarrier(3,new MeetingLeaderTask());
//        Executor executor=Executors.newFixedThreadPool(3);
        ExecutorService executor=Executors.newFixedThreadPool(3);
        executor.execute(new OpenMeetingTask(barrier,"C罗", 5));
        executor.execute(new OpenMeetingTask(barrier,"小罗", 3));
        executor.execute(new OpenMeetingTask(barrier,"卡卡", 1));
        executor.shutdown();
    

View Code

下面是代码原理 。主要讲解几个重要方法，还有成员变量的意义。

/**
 * 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
 * <em>cyclic</em> because it can be re-used after the waiting threads
 * are released.
 循环屏障是一个同步工具类，允许一系列的线程互相等待直到到达一个公岗的屏障点。
 栅栏在涉及固定大小的线程组必须偶尔互相等待的程序中很有用。
 屏障被叫做循环的是因为它可以在等待的线程被释放之后反复利用。
 *
 * <p>A <tt>CyclicBarrier</tt> supports an optional @link Runnable command
 * that is run once per barrier point, after the last thread in the party
 * arrives, but before any threads are released.
 * This <em>barrier action</em> is useful
 * for updating shared-state before any of the parties continue.
 *
 * <p><b>Sample usage:</b> Here is an example of
 *  using a barrier in a parallel decomposition design:
 * <pre>
 * class Solver 
 *   final int N;
 *   final float[][] data;
 *   final CyclicBarrier barrier;
 *
 *   class Worker implements Runnable 
 *     int myRow;
 *     Worker(int row)  myRow = row; 
 *     public void run() 
 *       while (!done()) 
 *         processRow(myRow);
 *
 *         try 
 *           barrier.await();
 *          catch (InterruptedException ex) 
 *           return;
 *          catch (BrokenBarrierException ex) 
 *           return;
 *         
 *       
 *     
 *   
 *
 *   public Solver(float[][] matrix) 
 *     data = matrix;
 *     N = matrix.length;
 *     barrier = new CyclicBarrier(N,
 *                                 new Runnable() 
 *                                   public void run() 
 *                                     mergeRows(...);
 *                                   
 *                                 );
 *     for (int i = 0; i < N; ++i)
 *       new Thread(new Worker(i)).start();
 *
 *     waitUntilDone();
 *   
 * 
 * </pre>
 * Here, each worker thread processes a row of the matrix then waits at the
 * barrier until all rows have been processed. When all rows are processed
 * the supplied @link Runnable barrier action is executed and merges the
 * rows. If the merger
 * determines that a solution has been found then <tt>done()</tt> will return
 * <tt>true</tt> and each worker will terminate.
 *
 * <p>If the barrier action does not rely on the parties being suspended when
 * it is executed, then any of the threads in the party could execute that
 * action when it is released. To facilitate this, each invocation of
 * @link #await returns the arrival index of that thread at the barrier.
 * You can then choose which thread should execute the barrier action, for
 * example:
 * <pre>  if (barrier.await() == 0) 
 *     // log the completion of this iteration
 *   </pre>
 *
 * <p>The <tt>CyclicBarrier</tt> uses an all-or-none breakage model
 * for failed synchronization attempts: If a thread leaves a barrier
 * point prematurely because of interruption, failure, or timeout, all
 * other threads waiting at that barrier point will also leave
 * abnormally via @link BrokenBarrierException (or
 * @link InterruptedException if they too were interrupted at about
 * the same time).
 *
 * <p>Memory consistency effects: Actions in a thread prior to calling
 * @code await()
 * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
 * actions that are part of the barrier action, which in turn
 * <i>happen-before</i> actions following a successful return from the
 * corresponding @code await() in other threads.
 *
 * @since 1.5
 * @see CountDownLatch
 *
 * @author Doug Lea
 */
 public class CyclicBarrier 
     /**
     * Each use of the barrier is represented as a generation instance.
     * The generation changes whenever the barrier is tripped, or
     * is reset. There can be many generations associated with threads
     * using the barrier - due to the non-deterministic way the lock
     * may be allocated to waiting threads - but only one of these
     * can be active at a time (the one to which <tt>count</tt> applies)
     * and all the rest are either broken or tripped.
     * There need not be an active generation if there has been a break
     * but no subsequent reset.
     每用一次屏障就意味着建立一个代的实例。
     世代会改变无论是被屏障阻塞或者被重置。
     在线程调用屏障时会有很多代关联。由于锁会以不确定的方式被分配来等待线程，但是只有这些中的一个在一次是激活的。并且所有剩下的不是在broken就是在阻塞。如果有一个break，没有后续重置，则不必有一个活跃的世代。
     
     
     */
    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
        condition会等待直到*/
    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;
     
/**
 * Waits until all @linkplain #getParties parties have invoked
 * <tt>await</tt> on this barrier.
 * 等待直到所有的线程都调用了这个屏障的await方法。
 * <p>If the current thread is not the last to arrive then it is
 * disabled for thread scheduling purposes and lies dormant until
 * one of the following things happens:
 如果当前线程不是最后一个到达的，那么它会为了线程调度和休眠变得不可用直到下面中的一个条件发生：
 * <ul>
 * <li>The last thread arrives; or
 * <li>Some other thread @linkplain Thread#interrupt interrupts
 * the current thread; or
 * <li>Some other thread @linkplain Thread#interrupt interrupts
 * one of the other waiting threads; or
 * <li>Some other thread times out while waiting for barrier; or
 * <li>Some other thread invokes @link #reset on this barrier.
 * </ul>
 *最后一个线程到达，或者其他线程打断当前线程；
 或者其他线程打断等待中的线程中的一个。
 或者其他线程在等待屏障的时候超时了。
 或者其他线程调用了屏障的reset方法。
 * <p>If the current thread:
 * <ul>
 * <li>has its interrupted status set on entry to this method; or
 * <li>is @linkplain Thread#interrupt interrupted while waiting
 * </ul>
 * then @link InterruptedException is thrown and the current thread‘s
 * interrupted status is cleared.
 *如果当前线程在进入这个方法时候有打断状态；或者当等待的时候被打断，则抛出InterruptedException
 中断异常，并且当前线程的中断状态被清空。
 * <p>If the barrier is @link #reset while any thread is waiting,
 * or if the barrier @linkplain #isBroken is broken when
 * <tt>await</tt> is invoked, or while any thread is waiting, then
 * @link BrokenBarrierException is thrown.
 *如果屏障当任何线程在等待的时候被重置，
 或者await方法被调用时，屏障被破坏掉。
 或者当任何线程等待时 会抛出中断异常。
 * <p>If any thread is @linkplain Thread#interrupt interrupted while waiting,
 * then all other waiting threads will throw
 * @link BrokenBarrierException and the barrier is placed in the broken
 * state.
 *如果任何线程在等待的时候被中断，所有其他等待的线程会抛出中断异常，并且屏障会变成broken状态。
 * <p>If the current thread is the last thread to arrive, and a
 * non-null barrier action was supplied in the constructor, then the
 * current thread runs the action before allowing the other threads to
 * continue.
 如果当前线程是最后一个到达的线程，并且一个非空的屏障操作在构造方法中被提供，
 那么在允许其他线程继续执行之前，当前线程会执行这个动作。
 * If an exception occurs during the barrier action then that exception
 * will be propagated in the current thread and the barrier is placed in
 * the broken state.
 *如果在屏障等待期间有异常发生，这个异常会在当前线程中传播，并且屏障会处于broken状态。
 * @return the arrival index of the current thread, where index
 *         <tt>@link #getParties() - 1</tt> indicates the first
 *         to arrive and zero indicates the last to arrive
 * @throws InterruptedException if the current thread was interrupted
 *         while waiting
 * @throws BrokenBarrierException if <em>another</em> thread was
 *         interrupted or timed out while the current thread was
 *         waiting, or the barrier was reset, or the barrier was
 *         broken when @code await was called, or the barrier
 *         action (if present) failed due an exception.
 */
public int await() throws InterruptedException, BrokenBarrierException 
    try 
        return dowait(false, 0L);
     catch (TimeoutException toe) 
        throw new Error(toe); // cannot happen;
    

/**
这个方法是比较重要的方法。 会涵盖各种的策略。
 * Main barrier code, covering the various policies.
 */
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();

        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();
                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)
                return index;

            if (timed && nanos <= 0L) 
                breakBarrier();
                throw new TimeoutException();
            
        
     finally 
        lock.unlock();
    

View Code

参考资料： https://blog.csdn.net/qq_38293564/article/details/80558157

https://www.jianshu.com/p/ddaecc027ba4

https://blog.csdn.net/qq_39241239/article/details/87030142

这个帖子讲的很通俗易懂。