【JUC源码解析】CyclicBarrier

简介

CyclicBarrier，一个同步器，允许多个线程相互等待，直到达到一个公共屏障点。

概述

CyclicBarrier支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后，释放所有线程之前，该命令只在屏障点运行一次。

应用

描述

有一个矩阵，每一行数据交给一个线程去处理，处理内容是，将这行数据的每一个值相加，结果存入第一个元素中，每个线程处理完成后，会在屏障点相互等待，直到最后一个线程也到达屏障点，最后将各个线程处理的数据汇总，具体是将每一行汇总的数据（存在每行的第一个元素中的数据）再相加，结果存在第一行中的第一个元素。

具体如下代码所示。

代码

public 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() {
            System.out.println("Matrix[" + myRow + "]数据准备处理");
            processRow(myRow); // 处理每一行数据，把各个数据相加，并且存入第一个元素
            System.out.println("Matrix[" + myRow + "]数据处理完成");
            try {
                barrier.await(); // 处理完成后，在此等待，直到最后一个线程也完成任务
            } catch (InterruptedException ex) {
                return;
            } catch (BrokenBarrierException ex) {
                return;
            }
        }
    }

    public Solver(float[][] matrix) throws InterruptedException {
        data = matrix;
        N = matrix.length;

        System.out.println("开始处理数据，最初矩阵如下所示");
        displayData(); // 数据展示

        Runnable barrierAction = new Runnable() { // 操作完成后，由最后一个到达屏障点的线程执行此操作，整体只执行一次
            public void run() { // 数据汇总，把第一列每行的和再相加，结果存入第一行第一个元素
                float tmp = 0.0f;
                for (int i = 0; i < N; i++) {
                    tmp += data[i][0];
                }

                data[0][0] = tmp;
                System.out.println("Matrix[0]数据汇总完成，结果是" + tmp + "，存在了Matrix[0][0]");
            }
        };
        barrier = new CyclicBarrier(N, barrierAction);

        List<Thread> threads = new ArrayList<Thread>(N);
        for (int i = 0; i < N; i++) {
            Thread thread = new Thread(new Worker(i));
            threads.add(thread);
            thread.start(); // 启动各个工作线程
        }

        for (Thread thread : threads) { // 等待所有线程执行完成
            thread.join();
        }
        System.out.println("所以数据都已经处理完成，最终矩阵如下所示");
        displayData();
    }

    private void processRow(int myRow) { //  处理每一行数据，把各个数据相加，并且存入第一个元素
        float[] row = data[myRow];
        float tmp = 0.0f;
        int length = row.length;
        for (int i = 0; i < length; i++) {
            tmp += row[i];
        }

        String msg = Arrays.toString(row);
        row[0] = tmp;
        System.out.println("Matrix[" + myRow + "]数据" + msg + "的和是：" + tmp + ", 且存入了Matrix[" + myRow + "][0]");
    }

    private void displayData() { // 数据展示
        TableInfo info = new TableInfo(N + 1); // TableInfo见下面代码，仅仅为了展示数据，可忽略，也可以用Arrays.toString(array);

        String[] header = new String[N + 1];
        header[0] = String.valueOf("    ");
        for (int i = 0; i < N; i++) {
            header[i + 1] = String.valueOf(i);
        }
        info.addHeader(header);
        for (int i = 0; i < N; i++) {
            String[] tmp = new String[N + 1];
            tmp[0] = String.valueOf(i);
            for (int j = 0; j < data[i].length; j++) {
                tmp[j + 1] = String.valueOf(data[i][j]);
            }
            info.addRecode(tmp);
        }
        System.out.println(info.getInfo());
    }

    public static void main(String[] args) throws InterruptedException {
        Random r = new Random(47);
        float[][] matrix = new float[5][5];
        for (int i = 0; i < 5; i++) {
            for (int j = 0; j < 5; j++) {
                matrix[i][j] = r.nextFloat();
            }
        }
        new Solver(matrix);
    }
}

class TableInfo {
    private int count;
    private int[] maxLens;
    private String[] columns;
    private List<String[]> records;
    private StringBuilder builder;

    public TableInfo(int count) {
        this.count = count;
        maxLens = new int[count];
        columns = new String[count];
        builder = new StringBuilder();
        records = new ArrayList<String[]>();
    }

    private boolean isValid(String... args) {
        return null == args || args.length == 0;
    }

    private boolean isOutOfIndex(String... args) {
        return args.length > count;
    }

    public boolean addHeader(String... record) {
        if (isValid(record) || isOutOfIndex(record)) {
            return false;
        }

        copy(columns, record);
        copy(maxLens, record);

        return false;
    }

    public boolean addRecode(String... record) {
        if (isValid(record) || isOutOfIndex(record)) {
            return false;
        }

        copy(maxLens, record);

        records.add(record);

        return false;
    }

    public String getInfo() {
        buildLine();
        buildHeader();
        buildLine();
        buildBody();
        buildLine();

        return builder.toString();
    }

    private void buildHeader() {
        builder.append("|");
        for (int i = 0; i < columns.length; i++) {
            builder.append(columns[i]);
            builder.append(getEmpty(maxLens[i] - columns[i].length()));
            builder.append("|");
        }
        builder.append("
");
    }

    private void buildBody() {
        for (String[] recode : records) {
            builder.append("|");
            for (int i = 0; i < recode.length; i++) {
                builder.append(recode[i]);
                builder.append(getEmpty(maxLens[i] - recode[i].length()));
                builder.append("|");
            }
            builder.append("
");
        }
    }

    private void buildLine() {
        builder.append("+");
        for (int i = 0; i < maxLens.length; i++) {
            builder.append(getLine(maxLens[i]));
            builder.append("+");
        }
        builder.append("
");
    }

    private String getLine(int count) {
        StringBuilder builder = new StringBuilder();
        for (int i = 0; i < count; i++) {
            builder.append("-");
        }

        return builder.toString();

    }

    private String getEmpty(int count) {
        StringBuilder builder = new StringBuilder();
        for (int i = 0; i < count; i++) {
            builder.append(" ");
        }

        return builder.toString();
    }

    private void copy(String[] args1, String... args2) {
        for (int i = 0; i < args2.length; i++) {
            args1[i] = args2[i];
        }
    }

    private void copy(int[] args1, String... args2) {
        for (int i = 0; i < args2.length; i++) {
            args1[i] = swap(args1[i], args2[i].length());
        }
    }

    private int swap(int max, int length) {
        if (max > length) {
            return max;
        }

        return length;
    }

输出

开始处理数据，最初矩阵如下所示
+----+----------+----------+----------+----------+----------+
|    |0         |1         |2         |3         |4         |
+----+----------+----------+----------+----------+----------+
|0   |0.72711575|0.39982635|0.5309454 |0.0534122 |0.16020656|
|1   |0.57799757|0.18847865|0.4170137 |0.51660204|0.73734957|
|2   |0.2678662 |0.9510573 |0.261361  |0.11435455|0.05086732|
|3   |0.5466897 |0.8037155 |0.20143336|0.76206654|0.55373144|
|4   |0.5304296 |0.15709275|0.5295954 |0.39661872|0.48718303|
+----+----------+----------+----------+----------+----------+

Matrix[0]数据准备处理
Matrix[1]数据准备处理
Matrix[0]数据[0.72711575, 0.39982635, 0.5309454, 0.0534122, 0.16020656]的和是：1.8715063, 且存入了Matrix[0][0]
Matrix[0]数据处理完成
Matrix[2]数据准备处理
Matrix[2]数据[0.2678662, 0.9510573, 0.261361, 0.11435455, 0.05086732]的和是：1.6455064, 且存入了Matrix[2][0]
Matrix[2]数据处理完成
Matrix[1]数据[0.57799757, 0.18847865, 0.4170137, 0.51660204, 0.73734957]的和是：2.4374416, 且存入了Matrix[1][0]
Matrix[1]数据处理完成
Matrix[4]数据准备处理
Matrix[3]数据准备处理
Matrix[3]数据[0.5466897, 0.8037155, 0.20143336, 0.76206654, 0.55373144]的和是：2.8676367, 且存入了Matrix[3][0]
Matrix[3]数据处理完成
Matrix[4]数据[0.5304296, 0.15709275, 0.5295954, 0.39661872, 0.48718303]的和是：2.1009195, 且存入了Matrix[4][0]
Matrix[4]数据处理完成
Matrix[0]数据汇总完成，结果是10.923011，存在了Matrix[0][0]
所以数据都已经处理完成，最终矩阵如下所示
+----+---------+----------+----------+----------+----------+
|    |0        |1         |2         |3         |4         |
+----+---------+----------+----------+----------+----------+
|0   |10.923011|0.39982635|0.5309454 |0.0534122 |0.16020656|
|1   |2.4374416|0.18847865|0.4170137 |0.51660204|0.73734957|
|2   |1.6455064|0.9510573 |0.261361  |0.11435455|0.05086732|
|3   |2.8676367|0.8037155 |0.20143336|0.76206654|0.55373144|
|4   |2.1009195|0.15709275|0.5295954 |0.39661872|0.48718303|
+----+---------+----------+----------+----------+----------+

源码解析

关于静态内部类Generation，其属性broken描述CyclicBarrier是否被打破。每次到达屏障点，或者重置时，都会新生下一代（创建Generation实例，记录下一批线程的屏障状态）。

一组线程，执行一批任务，调用CyclicBarrier的await方法，其内部调用的是Condition的await方法，实质上是调用LockSupport.park方法，入队等待。也就是说，这组线程一调用await方法，就会再此阻塞，而最后一个线程调用await方法时，不在走Condition的await的代码分支，而是先执行barrierCommand任务，然后调用nextGeneration方法，在该方法内部，会调用Condition的signalAll方法，即是唤醒阻塞在Condition上的线程，重置count，并重新创建Generation实例，一遍下次使用。

reset方法，会先打破原有的屏障，即是Generation的broken设置为true，提前调用Condition的signalAll方法，释放阻塞着的线程。并且接着调用nextGeneration，开启一个新的Generation实例。

问题：为什么要多次创建Generation实例呢？重用一个实例不是更好吗？毕竟只有一个broken属性，加一个set方法就好了？

回答：之所以创建新的Generation，是因为，每个Generation实例对应一批冲向屏障的线程。假如，只有一个Generation实例，想像这样一个场景，某个线程调用了reset方法，本意是使另外一组线程能重新使用CyclicBarrier，并唤醒与它同一批阻塞在Generation上的线程，被唤醒的老一批线程需要记录屏障打破记录（reset方法会打破屏障状态，即是broken为true），如果是同一个Generation对象，且broken为true，而新线程又要求broken为false，因为是全新的，对它们来说，屏障没有被打破。因此，无法满足。

以下是源码：

属性

    private static class Generation { // 分代，对应每一批冲向屏障的线程
        boolean broken = false; // 记录屏障是否被打破
    }

    private final ReentrantLock lock = new ReentrantLock(); // 可重入锁
    private final Condition trip = lock.newCondition(); // 条件
    private final int parties; // 记录线程数量
    private final Runnable barrierCommand; // 最后一个到达屏障的线程需要执行的任务
    private Generation generation = new Generation(); // 初代

    private int count; // 还未到达屏障的线程个数，会再次重置为parties

创建下一代

    private void nextGeneration() { // 创建下一代
        trip.signalAll(); // 唤醒阻塞在该代屏障上的线程
        count = parties; // 重置count
        generation = new Generation(); // 创建新的Generation实例
    }

打破屏障

    private void breakBarrier() { // 打破屏障
        generation.broken = true; // 设置broken
        count = parties; // 重置count
        trip.signalAll(); // 唤醒阻塞的线程
    }

关键的dowait

    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; // 来一个线程，count减1
            if (index == 0) { // 即将通过屏障
                boolean ranAction = false; // 记录是否正常完成
                try {
                    final Runnable command = barrierCommand; // 屏障点任务，由最后一个到达的线程负责执行
                    if (command != null)
                        command.run(); // 执行任务
                    ranAction = true; // 设置为正常完成
                    nextGeneration(); // 创建新的Generation对象
                    return 0; // 返回
                } finally {
                    if (!ranAction) // 如果没有正常完成，打破屏障
                        breakBarrier();
                }
            }

            for (;;) {
                try {
                    if (!timed) // 如果没有设置超时
                        trip.await(); // 调用await方法
                    else if (nanos > 0L) // 否则，调用awaitNanos方法
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) { // 如果是中断唤醒的，则看是否换代
                    if (g == generation && !g.broken) { // 如果还是同一代，并且屏障没有被打破，那么打破屏障，并抛出异常
                        breakBarrier();
                        throw ie;
                    } else { // 如果换代了，或者屏障已经打破了，什么都不作，仅仅重新设置中断标记
                        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(); // 解锁
        }
    }

行文至此结束。

尊重他人的劳动，转载请注明出处：http://www.cnblogs.com/aniao/p/aniao_cyclicbarrier.html