CyclicBarrier源码分析

CyclicBarrier是通过ReentrantLock(独占锁)和Condition来实现的。下面，我们分析CyclicBarrier中3个核心函数: 构造函数, await()作出分析。

1. 构造函数

CyclicBarrier的构造函数共2个：CyclicBarrier 和 CyclicBarrier(int parties, Runnable barrierAction)。第1个构造函数是调用第2个构造函数来实现的，下面第2个构造函数的源码。

public CyclicBarrier(int parties, Runnable barrierAction) {

    if (parties <= 0) throw new IllegalArgumentException();

    // parties表示“必须同时到达barrier的线程个数”。

    this.parties = parties;

    // count表示“处在等待状态的线程个数”。

    this.count = parties;

    // barrierCommand表示“parties个线程到达barrier时，会执行的动作”。

    this.barrierCommand = barrierAction;

}

2. 等待函数

CyclicBarrier.java中await()方法如下：

public int await() throws InterruptedException, BrokenBarrierException {

    try {

        return dowait(false, 0L);

    } catch (TimeoutException toe) {

        throw new Error(toe); // cannot happen;

    }

}

说明：await()是通过dowait()实现的。

private int dowait(boolean timed, long nanos)

    throws InterruptedException, BrokenBarrierException,

           TimeoutException {

    final ReentrantLock lock = this.lock;

    // 获取“独占锁(lock)”

    lock.lock();

    try {

        // 保存“当前的generation”

        final Generation g = generation;

        // 若“当前generation已损坏”，则抛出异常。

        if (g.broken)

            throw new BrokenBarrierException();

        // 如果当前线程被中断，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程。

        if (Thread.interrupted()) {

            breakBarrier();

            throw new InterruptedException();

        }

       // 将“count计数器”-1

       int index = --count;

       // 如果index=0，则意味着“有parties个线程到达barrier”。

       if (index == 0) {  // tripped

           boolean ranAction = false;

           try {

               // 如果barrierCommand不为null，则执行该动作。

               final Runnable command = barrierCommand;

               if (command != null)

                   command.run();

               ranAction = true;

               // 唤醒所有等待线程，并更新generation。

               nextGeneration();

               return 0;

           } finally {

               if (!ranAction)

                   breakBarrier();

           }

       }

        // 当前线程一直阻塞，直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，

        // 当前线程才继续执行。

        for (;;) {

            try {

                // 如果不是“超时等待”，则调用awati()进行等待；否则，调用awaitNanos()进行等待。

                if (!timed)

                    trip.await();

                else if (nanos > 0L)

                    nanos = trip.awaitNanos(nanos);

            } catch (InterruptedException ie) {

                // 如果等待过程中，线程被中断，则执行下面的函数。

                if (g == generation && ! g.broken) {

                    breakBarrier();

                    throw ie;

                } else {

                    Thread.currentThread().interrupt();

                }

            }

            // 如果“当前generation已经损坏”，则抛出异常。

            if (g.broken)

                throw new BrokenBarrierException();

            // 如果“generation已经换代”，则返回index。

            if (g != generation)

                return index;

            // 如果是“超时等待”，并且时间已到，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程。

            if (timed && nanos <= 0L) {

                breakBarrier();

                throw new TimeoutException();

            }

        }

    } finally {

        // 释放“独占锁(lock)”

        lock.unlock();

    }

}

说明：dowait()的作用就是让当前线程阻塞，直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，当前线程才继续执行。
(01) generation是CyclicBarrier的一个成员遍历，它的定义如下：

private Generation generation = new Generation();

private static class Generation {

    boolean broken = false;

}

在CyclicBarrier中，同一批的线程属于同一代，即同一个Generation；CyclicBarrier中通过generation对象，记录属于哪一代。
当有parties个线程到达barrier，generation就会被更新换代。

(02) 如果当前线程被中断，即Thread.interrupted()为true；则通过breakBarrier()终止CyclicBarrier。breakBarrier()的源码如下：

private void breakBarrier() {

    generation.broken = true;

    count = parties;

    trip.signalAll();

}

breakBarrier()会设置当前中断标记broken为true，意味着“将该Generation中断”；同时，设置count=parties，即重新初始化count；最后，通过signalAll()唤醒CyclicBarrier上所有的等待线程。

(03) 将“count计数器”-1，即--count；然后判断是不是“有parties个线程到达barrier”，即index是不是为0。
当index=0时，如果barrierCommand不为null，则执行该barrierCommand，barrierCommand就是我们创建CyclicBarrier时，传入的Runnable对象。然后，调用nextGeneration()进行换代工作，nextGeneration()的源码如下：

private void nextGeneration() {

    trip.signalAll();

    count = parties;

    generation = new Generation();

}

首先，它会调用signalAll()唤醒CyclicBarrier上所有的等待线程；接着，重新初始化count；最后，更新generation的值。

(04) 在for(;;)循环中。timed是用来表示当前是不是“超时等待”线程。如果不是，则通过trip.await()进行等待；否则，调用awaitNanos()进行超时等待。

原文地址：http://www.tuicool.com/articles/ERbIven

巴特西

CyclicBarrier源码分析

最新文章

热门文章