Java多线程再学习，温故知新（十三）CyclicBarrier

た入场券 2022-04-23 09:18 187阅读 0赞

## CyclicBarrier  ##

一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。

CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。

## 实现一个开会的例子，所有人到齐后开始开会 ##

import java.util.Random;
    import java.util.concurrent.CyclicBarrier;
    
    public class Demo {
    
    	Random random = new Random();
    
    	public void meeting(CyclicBarrier barrier) {
    		try {
    			Thread.sleep(random.nextInt(4000));
    		} catch (InterruptedException e1) {
    			e1.printStackTrace();
    		}
    		System.out.println(Thread.currentThread().getName() + " 到达会议室，等待开会..");
    
    		if(Thread.currentThread().getName().equals("Thread-7")) {
    			// Thread.currentThread().interrupt();
    			try {
    				Thread.sleep(10000);
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    			//barrier.reset();调用reset后线程将被唤醒
    			
    		}
    		
    		try {
    			barrier.await();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    
    	public static void main(String[] args) {
    		Demo demo = new Demo();
    
    		CyclicBarrier barrier = new CyclicBarrier(10, new Runnable() {
    			@Override
    			public void run() {
    				System.out.println("好！我们开始开会...");
    			}
    		});
    
    		for (int i = 0; i < 10; i++) {
    			new Thread(new Runnable() {
    				@Override
    				public void run() {
    					demo.meeting(barrier);
    				}
    			}).start();
    		}
    		
    		// 监控等待线程数
    		new Thread(new Runnable() {
    			
    			@Override
    			public void run() {
    				while(true) {
    					try {
    						Thread.sleep(1000);
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}
    					System.out.println("等待的线程数 " + barrier.getNumberWaiting());
    					System.out.println("is broken " + barrier.isBroken());
    				}
    			}
    		}).start();
    	}
    
    }

分析源码可以知道，CyclicBarrier底层是基于ReentrantLock和AbstractQueuedSynchronizer来实现的，所以，CyclicBarrier的数据结构也依托于AQS的数据结构

## 源码分析  ##

//基本属性
    public class CyclicBarrier {
        
        /** 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 */
        // 由最后一个进入 barrier 的线程执行的操作
        private final Runnable barrierCommand;
        /** The current generation */
        // 当前代，Generation类有一个属性broken，用来表示当前屏障是否被损坏
        private Generation generation = new Generation();
        // 正在等待进入屏障的线程数量
        private int count;
    }
    //构造函数之一
    public CyclicBarrier(int parties, Runnable barrierAction) {
            // 参与的线程数量小于等于0，抛出异常
            if (parties <= 0) throw new IllegalArgumentException();
            // 设置parties
            this.parties = parties;
            // 设置count
            this.count = parties;
            // 设置barrierCommand
            this.barrierCommand = barrierAction;
    }

## dowait函数 ##

此函数为CyclicBarrier类的核心函数，CyclicBarrier类对外提供的await函数在底层都是调用该了doawait函数

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) {  // 正在等待进入屏障的线程数量为0，所有线程都已经进入
                    // 运行的动作标识
                    boolean ranAction = false;
                    try {
                        // 保存运行动作
                        final Runnable command = barrierCommand;
                        if (command != null) // 动作不为空
                            // 运行
                            command.run();
                        // 设置ranAction状态
                        ranAction = true;
                        // 进入下一代
                        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) // 设置了等待时间，并且等待时间大于0
                            // 等待指定时长
                            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) { // 设置了等待时间，并且等待时间小于0
                        // 损坏屏障
                        breakBarrier();
                        // 抛出异常
                        throw new TimeoutException();
                    }
                }
            } finally {
                // 释放锁
                lock.unlock();
            }
        }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2llZmxleA_size_16_color_FFFFFF_t_70][]

## nextGeneration函数　 ##

此函数在所有线程进入屏障后会被调用，即生成下一个版本，所有线程又可以重新进入到屏障中，其源代码如下

private void nextGeneration() {
            // signal completion of last generation
            // 唤醒所有线程
            trip.signalAll();
            // set up next generation
            // 恢复正在等待进入屏障的线程数量
            count = parties;
            // 新生一代
            generation = new Generation();
        }

在此函数中会调用AQS的signalAll方法，即唤醒所有等待线程。如果所有的线程都在等待此条件，则唤醒所有线程。

public final void signalAll() {
                if (!isHeldExclusively()) // 不被当前线程独占，抛出异常
                    throw new IllegalMonitorStateException();
                // 保存condition队列头结点
                Node first = firstWaiter;
                if (first != null) // 头结点不为空
                    // 唤醒所有等待线程
                    doSignalAll(first);
            }
    //此函数判断头结点是否为空，即条件队列是否为空，然后会调用doSignalAll函数，doSignalAll函数源码如下　
    private void doSignalAll(Node first) {
                // condition队列的头结点尾结点都设置为空
                lastWaiter = firstWaiter = null;
                // 循环
                do {
                    // 获取first结点的nextWaiter域结点
                    Node next = first.nextWaiter;
                    // 设置first结点的nextWaiter域为空
                    first.nextWaiter = null;
                    // 将first结点从condition队列转移到sync队列
                    transferForSignal(first);
                    // 重新设置first
                    first = next;
                } while (first != null);
            }

此函数会依次将条件队列中的节点转移到同步队列中，会调用到transferForSignal函数，其源码如下

final boolean transferForSignal(Node node) {
            /*
             * If cannot change waitStatus, the node has been cancelled.
             */
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
    
            /*
             * Splice onto queue and try to set waitStatus of predecessor to
             * indicate that thread is (probably) waiting. If cancelled or
             * attempt to set waitStatus fails, wake up to resync (in which
             * case the waitStatus can be transiently and harmlessly wrong).
             */
            Node p = enq(node);
            int ws = p.waitStatus;
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                LockSupport.unpark(node.thread);
            return true;
        }

此函数的作用就是将处于条件队列中的节点转移到同步队列中，并设置结点的状态信息，其中会调用到enq函数，其源代码如下。

private Node enq(final Node node) {
            for (;;) { // 无限循环，确保结点能够成功入队列
                // 保存尾结点
                Node t = tail;
                if (t == null) { // 尾结点为空，即还没被初始化
                    if (compareAndSetHead(new Node())) // 头结点为空，并设置头结点为新生成的结点
                        tail = head; // 头结点与尾结点都指向同一个新生结点
                } else { // 尾结点不为空，即已经被初始化过
                    // 将node结点的prev域连接到尾结点
                    node.prev = t; 
                    if (compareAndSetTail(t, node)) { // 比较结点t是否为尾结点，若是则将尾结点设置为node
                        // 设置尾结点的next域为node
                        t.next = node; 
                        return t; // 返回尾结点
                    }
                }
            }
        }

breakBarrier函数

此函数的作用是损坏当前屏障，会唤醒所有在屏障中的线程。源代码如下

private void breakBarrier() {
            // 设置状态
            generation.broken = true;
            // 恢复正在等待进入屏障的线程数量
            count = parties;
            // 唤醒所有线程
            trip.signalAll();
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2llZmxleA_size_16_color_FFFFFF_t_70]: /images/20220321/0afc05f1d5324f41b8d87740228bc793.png