十：深入理解 CyclicBarrier—— 栅栏锁

短命女 2024-05-05 07:45 6阅读 0赞

#### 目录 ####

*  1、CyclicBarrier 入门
 *   *  1.1、概念
     *  1.2、案例
 *  2、CyclicBarrier 源码分析
 *   *  2.1、类结构
     *  2.2、\`await()\` 方法 —— CyclicBarrier
     *   *  2.2.1、\`dowait()\` 方法 —— CyclicBarrier
         *   *  2.2.1.1、\`breakBarrier()\` 方法 —— CyclicBarrier
             *  2.2.1.2、\`nextGeneration()\` 方法 —— CyclicBarrier
 *  3、CyclicBarrier 与 CountDownLatch 区别

## 1、CyclicBarrier 入门 ##

### 1.1、概念 ###

`CyclicBarrier`：让一组线程到达某个屏障，被阻塞，一直到组内的最后一个线程到达，然后屏障开放，接着，所有的线程继续运行

`CyclicBarrier` 是通过独占锁实现的，底层包含了 “ReentrantLock 对象 lock” 和 “Condition 对象 trip”，通过条件队列 trip 来对线程进行阻塞的，并且其内部维护了两个 int 型的变量 parties 和 count

### 1.2、案例 ###

public static void main(String[] args) throws InterruptedException {
        
        final int threadCount = 5;
        Runnable task = () -> System.out.println("朋友集合完毕，准备出发");
        CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount, new Thread(task));
        for (int i = 0; i < threadCount; i++) {
        
            new Thread(() -> {
        
                try {
        
                    System.out.println(Thread.currentThread().getName() + " 到了");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName() + " 已到");
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
            }, "朋友" + i).start();
        }
    }

执行结果

朋友2 到了
    朋友3 到了
    朋友0 到了
    朋友4 到了
    朋友1 到了
    朋友集合完毕，准备出发
    朋友1 已到
    朋友2 已到
    朋友3 已到
    朋友0 已到
    朋友4 已到

## 2、CyclicBarrier 源码分析 ##

### 2.1、类结构 ###

public class CyclicBarrier {
        
    	// 锁实例
    	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();
    	// 计数器【阻塞线程数】：它的初始值和 parties 相同，以后随着每次 await 方法的调用而减 1，直到减为 0 就将所有线程唤醒
    	private int count;
    	
    	public CyclicBarrier(int parties) {
        
            this(parties, null);
        }
    	
    	public CyclicBarrier(int parties, Runnable barrierAction) {
        
            if (parties <= 0){
        
            	throw new IllegalArgumentException();
           	}
            this.parties = parties;
            this.count = parties;
            this.barrierCommand = barrierAction;
        }
        
        // 内部类：代表栅栏的当前代，利用它可以实现循环等待，当 count 减为 0 会将所有阻塞的线程唤醒，并设置成下一代
        private static class Generation {
        
        	// 栅栏是否被破坏
            boolean broken = false;
        }
    }

查看构造方法：

1.  第一个参数也是 Int 类型的，传入的是执行线程的个数。这个数量和 `CountDownLatch` 不一样，这个数量是需要和线程数量相同的；而 `CountDownLatch` 可以大于等于
2.  第二个参数是 barrierAction，这个参数是当屏障开放后，执行的任务线程。如果当屏障开放后需要执行什么任务，可以写在这个线程中

如下图：

![在这里插入图片描述][fc4bc64893e84210b7c1efe17a1b3899.png]

### 2.2、`await()` 方法 —— CyclicBarrier ###

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

#### 2.2.1、`dowait()` 方法 —— CyclicBarrier ####

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()) {
        
            	// 如果线程被中断，将栅栏设置为破坏状态【generation.broken = true;】，且唤醒所有阻塞的线程
                breakBarrier();
                // 抛异常
                throw new InterruptedException();
            }
            // 已加锁，线程安全
            int index = --count;
            if (index == 0) {
        
            	// 计数器的值减为 0 ，则需唤醒所有线程并转换到下一代
                boolean ranAction = false;
                try {
        
                	// 先执行指定的任务【并未开启新线程：谁是最后一个线程谁执行】
                    final Runnable command = barrierCommand;
                    if (command != null) {
        
                    	command.run();
                   	}
                    ranAction = true;
                    // 唤醒所有线程并转到下一代
                    nextGeneration();
                    return 0;
                } finally {
        
                    if (!ranAction) {
        
                    	// 没有执行的话，再次执行
                    	breakBarrier();
                   	}
                }
            }
            // 一直自旋，直到换代、被破坏、中断、超时
            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 {
        
                    	// 若在捕获中断异常前已经完成在栅栏上的等待, 则直接调用中断操作
                        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();
        }
    }

##### 2.2.1.1、`breakBarrier()` 方法 —— CyclicBarrier #####

private void breakBarrier() {
        
        generation.broken = true;
        count = parties;
        // 唤醒所有线程
        trip.signalAll();
    }

##### 2.2.1.2、`nextGeneration()` 方法 —— CyclicBarrier #####

private void nextGeneration() {
        
    	// 唤醒所有线程
        trip.signalAll();
        count = parties;
        // 设置下一代
        generation = new Generation();
    }

## 3、CyclicBarrier 与 CountDownLatch 区别 ##

1.  `CountDownLatch` 参与的线程分为两类：一个是等待者，另一个是计数者；`CyclicBarrier` 参与的线程既是等待者，也是计数者
2.  `CountDownLatch` 完成一次完整的协作过程后不能再复用；`CyclicBarrier` 可以复用
3.  `CountDownLatch` 的计数值与大于等于线程数量；`CyclicBarrier` 的初始计数值与线程个数一致
4.  `CountDownLatch` 基于AQS 的 state 实现；`CyclicBarrier` 基于 `ReentrantLock&Condition` 实现

[fc4bc64893e84210b7c1efe17a1b3899.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/71442dd9affb44e9bccf746b9b60674e.png