JAVA多线程-CyclicBarrier同步屏障

一、概述

CyclicBarrier，也是位于java.util.concurrent包下的一个并发同步工具类，光从字面上理解的意思是可循环使用（Cyclic）的屏障（Barrier）。

主要的作用是：可以让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到这组线程中的最后一个线程到达屏障时，屏障才会被移除，然后就可以使得所有被屏障拦截的线程能够继续往下干活。我们暂且把这个状态就叫做barrier，当调用await()方法之后，线程就处于barrier了。

二、应用场景

CyclicBarrier可以用于多线程计算数据，最后合并计算结果的应用场景。比如我们用一个Excel文件保存了用户的所有银行流水，然后每个sheet工作簿保存一个账户最近一年的每笔流水记录，现在需要统计用户的日均银行流水，那么就可以使用多线程处理每个sheet工作簿里的银行流水，全部执行完之后，得到每个sheet的日均银行流水，最后，再一起统计这些线程的计算结果，从而最终计算出整个Excel的日均银行流水。

三、使用方法

3.1、默认构造器

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

public CyclicBarrier(int parties) {
    this(parties, null);
}

其中，parties这个参数表示屏障拦截的线程数量，每个线程调用await()方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞，而barrierAction这个参数的作用则是：当这些线程都到达了屏障时，可以额外做某些事情的，其本身是传入了一个Runnable类型的参数，然后可以在run()方法中可以额外做一些事情。

3.2、重要方法

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

通常比较常用，可以用于挂起一组线程，直至所有的线程都到达屏障了再同时执行后续任务。

public int await(long timeout, TimeUnit unit)
        throws InterruptedException,
               BrokenBarrierException,
               TimeoutException {
    return dowait(true, unit.toNanos(timeout));
}

可以让这些那些先到达屏障状态的线程等待一定的时间，如果此时还有线程没有到达屏障则直接让已经到达屏障的线程继续执行后续任务。

四、代码实例

1. 假若有若干个线程都要进行写数据操作，并且只有所有线程都完成写数据操作之后，这些线程才能继续做后面的事情，此时就可以利用CyclicBarrier了。

package com.feizi.java.concurrency.tool;

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

/**
 * Created by feizi on 2018/6/4.
 */
public class TestCyclicBarrier {
    public static void main(String[] args) {
        int N = 4;
        CyclicBarrier barrier = new CyclicBarrier(N);
        for (int i = 0; i < N; i++){
            new Writer(barrier).start();
        }
    }

    static class Writer extends Thread{
        private CyclicBarrier barrier;

        public Writer(CyclicBarrier barrier) {
            this.barrier = barrier;
        }

        @Override
        public void run() {
            System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
            try {
                //模拟写入耗时操作
                Thread.sleep(5000);
                System.out.println("线程" + Thread.currentThread().getName() + "写入数据完毕，等待其他线程写入完毕...");

                barrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println("所有线程写入完毕，继续处理其他任务...");
        }
    }
}

控制台输出结果：

线程Thread-1正在写入数据...
线程Thread-0正在写入数据...
线程Thread-3正在写入数据...
线程Thread-2正在写入数据...
线程Thread-1写入数据完毕，等待其他线程写入完毕...
线程Thread-2写入数据完毕，等待其他线程写入完毕...
线程Thread-3写入数据完毕，等待其他线程写入完毕...
线程Thread-0写入数据完毕，等待其他线程写入完毕...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...

Process finished with exit code 0

从上面输出结果可以看到，每个线程执行完毕之后，都在等待其他线程执行完毕，当所有线程都执行完毕之后，则他们就可以继续进行后续的任务了。

2. 如果想在所有线程都到达屏障后，还需要进行一些额外的操作，那么这个时候可以使用上面说的第二个构造函数了，传入一个Runnable参数进行处理。

package com.feizi.java.concurrency.tool;

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

/**
 * Created by feizi on 2018/6/5.
 */
public class TestCyclicBarrierDemo {
    public static void main(String[] args) {
        int N = 4;

        CyclicBarrier barrier = new CyclicBarrier(N, new Runnable() {
            @Override
            public void run() {
                System.out.println("当前线程" + Thread.currentThread().getName() + "被翻牌了...");
            }
        });

        for (int i = 0; i < N; i++){
            new Writer(barrier).start();
        }
    }

    static class Writer extends Thread{
        private CyclicBarrier barrier;

        public Writer(CyclicBarrier barrier) {
            this.barrier = barrier;
        }

        @Override
        public void run() {
            System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
            try {
                //模拟耗时操作
                Thread.sleep(5000);

                System.out.println("线程" + Thread.currentThread().getName() + "写入数据完毕...");
                barrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println("所有线程写入完毕，继续处理其他任务...");
        }
    }
}

控制台输出结果：

线程Thread-0正在写入数据...
线程Thread-1正在写入数据...
线程Thread-2正在写入数据...
线程Thread-3正在写入数据...
线程Thread-1写入数据完毕...
线程Thread-2写入数据完毕...
线程Thread-3写入数据完毕...
线程Thread-0写入数据完毕...
当前线程Thread-0被翻牌了...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...
所有线程写入完毕，继续处理其他任务...

Process finished with exit code 0

从上面的输出结果我们可以看到，当4个线程都到达屏障后，CyclicBarrier会从这组线程中选择一个线程（随机选择，跟CPU调度有关，谁被分配了时间片，就谁去执行）去执行Runnable中的内容，也就是额外的操作了。我们看到上面选择的是Thread-0这个线程。

3. 如果想让那些先到达屏障状态的线程等待一定的时间，然后如果此时还有线程没有到达屏障则直接让已经到达屏障的线程继续执行后续任务。那么我们可以使用上面介绍的这个public int await(long timeout, TimeUnit unit)方法。

package com.feizi.java.concurrency.tool;

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

/**
 * Created by feizi on 2018/6/5.
 */
public class TestCyclicBarrierDemo1 {

    public static void main(String[] args) {
        int N = 4;
        CyclicBarrier barrier = new CyclicBarrier(N);

        for (int i = 0; i < N; i++){
            if(i < N -1){
                new Writer(barrier).start();
            }else {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                new Writer(barrier).start();
            }
        }
    }

    static class Writer extends Thread{
        private CyclicBarrier barrier;

        public Writer(CyclicBarrier barrier) {
            this.barrier = barrier;
        }

        @Override
        public void run() {
            System.out.println("线程 " + Thread.currentThread().getName() + "正在写入数据...");

            try {
                //模拟耗时操作
                Thread.sleep(1000);

                System.out.println("线程 " + Thread.currentThread().getName() + "写入数据完毕， 等待其他线程写入完毕...");

                try {
                    barrier.await(5000, TimeUnit.MILLISECONDS);
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println(Thread.currentThread().getName() + ", 所有线程写入完毕，继续处理其他任务...");
        }
    }
}

控制台输出结果：

线程 Thread-0正在写入数据...
线程 Thread-2正在写入数据...
线程 Thread-1正在写入数据...
线程 Thread-0写入数据完毕， 等待其他线程写入完毕...
线程 Thread-2写入数据完毕， 等待其他线程写入完毕...
线程 Thread-1写入数据完毕， 等待其他线程写入完毕...
线程 Thread-3正在写入数据...
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:250)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:435)
    at com.feizi.java.concurrency.tool.TestCyclicBarrierDemo1$Writer.run(TestCyclicBarrierDemo1.java:49)
java.util.concurrent.TimeoutException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:257)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:435)
    at com.feizi.java.concurrency.tool.TestCyclicBarrierDemo1$Writer.run(TestCyclicBarrierDemo1.java:49)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:250)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:435)
    at com.feizi.java.concurrency.tool.TestCyclicBarrierDemo1$Writer.run(TestCyclicBarrierDemo1.java:49)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:207)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:435)
    at com.feizi.java.concurrency.tool.TestCyclicBarrierDemo1$Writer.run(TestCyclicBarrierDemo1.java:49)
线程 Thread-3写入数据完毕， 等待其他线程写入完毕...
Thread-2, 所有线程写入完毕，继续处理其他任务...
Thread-0, 所有线程写入完毕，继续处理其他任务...
Thread-1, 所有线程写入完毕，继续处理其他任务...
Thread-3, 所有线程写入完毕，继续处理其他任务...

Process finished with exit code 0

从上面输出结果，我们可以看到，因为在main方法测试的时候，我们特意指定了让最后一个线程延迟启动，由于在前面三个线程都到达了屏障之后，在等待了指定的时间之后发现第4个线程还没有到达屏障，然后就会抛出异常而去继续执行后面的任务了。

五、CyclicBarrier和CountDownLatch的区别

1. CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景，比如如果计算发生错误，可以重置计数器，并让线程们重新执行一次。

2. CyclicBarrier还提供其他有用的方法，比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken方法用来知道阻塞的线程是否被中断。

除此之外，CyclicBarrier还可以进行重用，即重复使用：

package com.feizi.java.concurrency.tool;

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

/**
 * Created by feizi on 2018/6/5.
 */
public class TestCyclicBarrierDemo2 {
    public static void main(String[] args) {
        int N = 4;

        CyclicBarrier barrier = new CyclicBarrier(N);

        for (int i = 0; i < N; i++){
            new Writer(barrier).start();
        }

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("CyclicBarrierz重用1...");

        for (int i = 0; i < N; i++){
            new Writer(barrier).start();
        }

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("CyclicBarrierz重用2...");

        for (int i = 0; i < N; i++){
            new Writer(barrier).start();
        }
    }

    static class Writer extends Thread{
        private CyclicBarrier barrier;

        public Writer(CyclicBarrier barrier) {
            this.barrier = barrier;
        }

        @Override
        public void run() {
            System.out.println("线程 " + Thread.currentThread().getName() + " 正在写入数据...");
            try {
                Thread.sleep(5000);

                System.out.println("线程 " + Thread.currentThread().getName() + " 写入数据完毕，等待其他线程写入完毕...");
                barrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }

            System.out.println(Thread.currentThread().getName() + "， 所有线程写入完毕，继续处理其他任务...");
        }
    }
}

控制台输出结果：

线程 Thread-0 正在写入数据...
线程 Thread-1 正在写入数据...
线程 Thread-2 正在写入数据...
线程 Thread-3 正在写入数据...
线程 Thread-0 写入数据完毕，等待其他线程写入完毕...
线程 Thread-3 写入数据完毕，等待其他线程写入完毕...
线程 Thread-1 写入数据完毕，等待其他线程写入完毕...
线程 Thread-2 写入数据完毕，等待其他线程写入完毕...
Thread-3， 所有线程写入完毕，继续处理其他任务...
Thread-0， 所有线程写入完毕，继续处理其他任务...
Thread-2， 所有线程写入完毕，继续处理其他任务...
Thread-1， 所有线程写入完毕，继续处理其他任务...
CyclicBarrierz重用1...
线程 Thread-4 正在写入数据...
线程 Thread-5 正在写入数据...
线程 Thread-6 正在写入数据...
线程 Thread-7 正在写入数据...
线程 Thread-5 写入数据完毕，等待其他线程写入完毕...
线程 Thread-4 写入数据完毕，等待其他线程写入完毕...
线程 Thread-6 写入数据完毕，等待其他线程写入完毕...
线程 Thread-7 写入数据完毕，等待其他线程写入完毕...
Thread-5， 所有线程写入完毕，继续处理其他任务...
Thread-6， 所有线程写入完毕，继续处理其他任务...
Thread-4， 所有线程写入完毕，继续处理其他任务...
Thread-7， 所有线程写入完毕，继续处理其他任务...
CyclicBarrierz重用2...
线程 Thread-8 正在写入数据...
线程 Thread-9 正在写入数据...
线程 Thread-10 正在写入数据...
线程 Thread-11 正在写入数据...
线程 Thread-9 写入数据完毕，等待其他线程写入完毕...
线程 Thread-10 写入数据完毕，等待其他线程写入完毕...
线程 Thread-8 写入数据完毕，等待其他线程写入完毕...
线程 Thread-11 写入数据完毕，等待其他线程写入完毕...
Thread-11， 所有线程写入完毕，继续处理其他任务...
Thread-8， 所有线程写入完毕，继续处理其他任务...
Thread-10， 所有线程写入完毕，继续处理其他任务...
Thread-9， 所有线程写入完毕，继续处理其他任务...

Process finished with exit code 0

从上面输出结果，我们可以看到，在首次的4个线程穿过屏障之后，又可以继续重新一轮的使用，而CountDownLatch则无法做到重复使用。

六、原文参考

1. http://ifeve.com/concurrency-cyclicbarrier/
2. http://www.importnew.com/21889.html