源码剖析之CyclicBarrier

CyclicBarrier：jdk current 包提供了一个让多个线程在某个点到达之前都互相等待的工具类，并且可以多次循环使用，故曰：循环障碍器。

使用场景：它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。
CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在屏障点运行一次，（注意：实际上是在最后一个线程中执行的！！！）若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。 【···引用jdk中文翻译】



实现依赖：
1、lock 可重入锁 ReentrantLock
2、Condition 条件谓词。

实现思路：就是设置一个计数，每当有线程达到时，计数count-1，Condition.await 进入阻塞， 当count =0，那么可以 signalAll ，让所有线程得以唤醒。 唤醒后立马重置！

核心代码分析如下：

public class CyclicBarrier {
    /** 进入障碍前必须获取的锁 */
    private final ReentrantLock lock = new ReentrantLock();
    /** 用来表示阻塞的条件 */
    private final Condition trip = lock.newCondition();
    /** 障碍释放前需要wait的线程数量*/
    private final int parties;
    /* 障碍突破后 要执行的命令 */
    private final Runnable barrierCommand;
    /** 英文翻译：代。 其实更准确的含义是 轮，相当于一次集合到释放为为一轮，一轮一轮的进行*/
    private Generation generation = new Generation();

    /**
     重置后count = parties；表示：目前等待还需要的线程的数量才能结束当前轮，进入下一轮
     */
    private int count;

    
/** 跳出障碍后，唤醒所有等待中的线程， 重置障碍器状态 （获取锁后才可以调用）
注意：generation  被被改变，那么其他线程会优先被 signalAll 。(分析dowait 请结合此处的逻辑)*/
    private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation();
    }

/* 打破障碍，设置当前线程的generation ，并唤醒所有等待中的线程
注意：只被这个方法调用才有机会 设置generation.broken = true ，这个方法调用后 其他等待的线程是会被trip.signalAll()的 ，如果没有进入下一轮，后来的线程是直接抛出异常的。(分析dowait 请结合此处的逻辑)
*/
 private void breakBarrier() {
        generation.broken = true;
	count = parties;
        trip.signalAll();
    }

/** 
parties - 在启动 barrier 前必须调用 await() 的线程数
barrierAction - 在启动 barrier 时执行的命令；如果不执行任何操作，则该参数为 null 
*/
 public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }


/**
在所有参与者中的最后一个线程到达之前，所有已经在此 barrier 上调用 await 方法都将进入阻塞状态！
如果当前线程不是将到达的最后一个线程，出于调度目的，将阻塞它，且在发生以下情况之一前，该线程将一直处于休眠状态： 

1、最后一个线程到达；（很容易理解吧：nextGeneration 和 breakBarrier 方法 无论调用哪一个都会signalAll 的）
2、其他某个线程中断当前线程；（处于阻塞于此障碍器）另一个等待线程；（当前状态 wait，那么其他线程中断当前线程，wait 会抛出异常（如果还不明白，请仔细理解Condition.wait 的含义））
3、其他某个线程中断（其他线程被中断了，那么其他线程会调用breakBarrier 方法，然后notifyALl）
4、其他某个线程在等待 barrier 时超时；(效果同上：会调用breakBarrier)
5、其他某个线程在此 barrier 上调用 reset()。 （同3）

如果当前线程： 
1、在进入此方法时已经设置了该线程的中断状态； 
2、在等待时被中断  
则抛出 InterruptedException，并且清除当前线程的已中断状态。 

*/
 public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen;
        }
    }


 /** 线程wait的真实实现
 timed：是否是有时间限制
 nanos:wait的纳秒数
 */
 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) //如果当前状态已经break ,如果breakBarrier 方法的调用没有和nextGeneration 同时出现，那么后继的线程会在这里抛出异常！！！
                throw new BrokenBarrierException();

            if (Thread.interrupted()) { //如果当前线程已经中断，那么break 障碍器，释放已经在此等待的线程，并抛出异常。 如果此后还有线程跟进，那么会直接抛出异常，其实就是上面的这句：if(g.broken) !!!
                breakBarrier();
                throw new InterruptedException();
            }

           int index = --count; //随着当前线程的道理，如果index == 0 ，那么当前线程就是最后一个到达的线程，即：会引起障碍器打破
           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(); //最后无论如何，打破障碍器！ 主要是针对：command.run 可以抛出异常的情况！！！
               }
           }

            // 循环直到 障碍逃出 打破 中断 或者超时！
            for (;;) {
                try {
                    if (!timed) //如果不支持超时，那么直接在条件谓词处等待！
                        trip.await();
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos); //如果是nanos >0,那么在trip上等待nanos时间！！
                } catch (InterruptedException ie) { //如果wait得到notify，那么肯定不会进入这块！！！只有当前线程被打断 才有机会。
                    if (g == generation && ! g.broken) { //只有其他线程打断当前线程 会进入这块
                        breakBarrier(); //如果发生异常，那么打破障碍器，并抛出异常！！！
			                  throw ie;
							    } else {
	/*这个是个十分特殊的地方经过仔细的思考，我发现这块代码压根走不到！！！ 进入此处，那么一定其他线程中断了当前线程，如果wait状态下没有被notifyAll唤醒，那么g == generation 必然成立，而generation.broken = true 时，可以肯定会被唤醒而不是中断当前线程。（这块理解起来确实比较别扭）	

 private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation(); //如果当前的generation 变了，那么会先被signalAll，不会进入这里
    }

*/			Thread.currentThread().interrupt();
							    }
                }

   /**如果g已经被打破了，那么抛出异常
   想想什么情况下g.broken 会为false 呢？
   */
                if (g.broken) 
                    throw new BrokenBarrierException();

                if (g != generation) //如果g != geneation ，那么可以认定：障碍器重置了，或者所有线程已经达到了！！
                    return index;

                if (timed && nanos <= 0L) { //如果是支持超时，并且nanos<=0 ,那么打破障碍器，并抛出超时的异常
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock(); //最后不要忘记解锁
        }
    }







/** 重置障碍器*/
public void reset() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            breakBarrier();   // break the current generation
            nextGeneration(); // start a new generation
        } finally {
            lock.unlock();
        }
    }

/** 
返回当前在屏障处等待的参与者数目
*/
 public int getNumberWaiting() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return parties - count;
        } finally {
            lock.unlock();
        }
    }

}

以上是我个人的分析和理解，如果逻辑偏差，请指出，非常感谢！

评论

1 楼 lvwenwen 2013-06-07  

源码剖析之CyclicBarrier