CyclicBarrier简介

简介

CyclicBarrier也是java并发包中的一个工具类，也被称为同步屏障，作用是可以让一组线程达到一个屏障时阻塞，直到最后一个线程到达屏障时，所有被阻塞的线程才能继续执行。
示意图如下：

下面将从源码的角度来看看CyclicBarrier并发类，如无特殊说明，本文JDK版本为1.8

构造方法

该类中有2个构造方法，分别来看下：

参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier已经到达屏障位置，线程被阻塞

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

比第一个构造方法多了一个参数barrierAction，表示在所有线程到达屏障后执行

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

实现原理

CyclicBarrier实现主要基于ReentrantLock，本文就不介绍了ReentrantLock，不清楚以的[请点击](https://starlin.top/2018/07/09/java%E5%B9%B6%E5%8F%91/java
%E5%B9%B6%E5%8F%91%E4%B9%8BReentrantLock/)了解详情，

     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 */
    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();

await()实现

内部其实是调用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;
            if (index == 0) {  // tripped ，说明所有线程已经到达屏障处
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    // 该方法中执行signalAll()唤醒其他等待线程
                    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();
        }
    }

大概过程如下：

每当线程执行await，内部变量count减1，如果count不等于0，说明有线程未到达屏障处，则在锁条件变量trip上等待
当count等于0，说明所有线程都到达屏障处，开始执行signalAll方法，唤醒其他等待线程，其中nextGeneration()可以实现屏障的循环使用

重新生成Generation对象
恢复count值

 private void nextGeneration() {
    // signal completion of last generation
    trip.signalAll();
    // set up next generation
    count = parties;
    generation = new Generation();
}

应用场景

假设某一个部门团建，要求每一位员工上午11点集合，统一出发
在这个例子中将部门作为一个主线程，每位员工作为一个子线程，Demo如下：

 public class CyclicBarrierComPanyTest {
    public static void main(String[] args) throws InterruptedException {
        // 表示人数
        int count = 10;
        CyclicBarrier cyclicBarrier = new CyclicBarrier(count + 1);
 
        // 创建线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(count);
 
        for (int i = 0; i < count; i++) {
            // 将子线程加入到主线程中
            threadPool.execute(new Employee(cyclicBarrier,i + 1));
            Thread.sleep(10);
        }
        try {
            // 阻塞线程，直到所有人到达集合点才执行（出发去目的地）
            cyclicBarrier.await();
            Thread.sleep(10);
            System.out.println("所有人已经到达集合节点，开始前往目的地");
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }finally {
            threadPool.shutdown();
        }
    }
}
 
// 每个员工代表子线程
class Employee implements Runnable {
 
    private CyclicBarrier cyclicBarrier;
 
    private int employeeIndex;
 
    public Employee(CyclicBarrier barrier,int employeeIndex){
        this.cyclicBarrier = barrier;
        this.employeeIndex = employeeIndex;
    }
 
    @Override
    public void run() {
        try {
            System.out.println("员工：" + employeeIndex + ",正在前往集合");
            Thread.sleep(10*employeeIndex);
            System.out.println("员工：" + employeeIndex + ",已到达集合点");
            cyclicBarrier.await();
            Thread.sleep(10);
            System.out.println("员工：" + employeeIndex + ",已出发前往目的地");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

运行可能结果如下：

员工：1,正在前往集合
员工：2,正在前往集合
员工：1,已到达集合点
员工：2,已到达集合点
员工：3,正在前往集合
员工：5,正在前往集合
员工：4,正在前往集合
员工：6,正在前往集合
员工：7,正在前往集合
员工：8,正在前往集合
员工：3,已到达集合点
员工：4,已到达集合点
员工：9,正在前往集合
员工：10,正在前往集合
员工：5,已到达集合点
员工：6,已到达集合点
员工：7,已到达集合点
员工：8,已到达集合点
员工：9,已到达集合点
员工：10,已到达集合点
员工：8,已出发前往目的地
员工：5,已出发前往目的地
员工：3,已出发前往目的地
员工：2,已出发前往目的地
员工：1,已出发前往目的地
员工：6,已出发前往目的地
员工：7,已出发前往目的地
员工：9,已出发前往目的地
员工：4,已出发前往目的地
员工：10,已出发前往目的地

与CountDownLatch的区别

CyclicBarrier一般用于一组线程互相等待至某个状态，然后这一组线程再同时执行；CountDownLatch一般用于某个线程A等待若干个其他线程执行完任务之后，它才执行；
CyclicBarrier可以循环使用，而CountDownLatch不是循环使用的

end，感谢阅读！！！

# java并发

简介

构造方法

实现原理

await()实现

应用场景

与CountDownLatch的区别

CountDownLatch介绍

布隆过滤器浅析

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

	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 */
	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();

	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;
	// 该方法中执行signalAll()唤醒其他等待线程
	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();
	}
	}

	private void nextGeneration() {
	// signal completion of last generation
	trip.signalAll();
	// set up next generation
	count = parties;
	generation = new Generation();
	}

	public class CyclicBarrierComPanyTest {
	public static void main(String[] args) throws InterruptedException {
	// 表示人数
	int count = 10;
	CyclicBarrier cyclicBarrier = new CyclicBarrier(count + 1);

	// 创建线程池
	ExecutorService threadPool = Executors.newFixedThreadPool(count);

	for (int i = 0; i < count; i++) {
	// 将子线程加入到主线程中
	threadPool.execute(new Employee(cyclicBarrier,i + 1));
	Thread.sleep(10);
	}
	try {
	// 阻塞线程，直到所有人到达集合点才执行（出发去目的地）
	cyclicBarrier.await();
	Thread.sleep(10);
	System.out.println("所有人已经到达集合节点，开始前往目的地");
	} catch (BrokenBarrierException e) {
	e.printStackTrace();
	}finally {
	threadPool.shutdown();
	}
	}
	}

	// 每个员工代表子线程
	class Employee implements Runnable {

	private CyclicBarrier cyclicBarrier;

	private int employeeIndex;

	public Employee(CyclicBarrier barrier,int employeeIndex){
	this.cyclicBarrier = barrier;
	this.employeeIndex = employeeIndex;
	}

	@Override
	public void run() {
	try {
	System.out.println("员工：" + employeeIndex + ",正在前往集合");
	Thread.sleep(10*employeeIndex);
	System.out.println("员工：" + employeeIndex + ",已到达集合点");
	cyclicBarrier.await();
	Thread.sleep(10);
	System.out.println("员工：" + employeeIndex + ",已出发前往目的地");
	} catch (InterruptedException \| BrokenBarrierException e) {
	e.printStackTrace();
	}
	}
	}