CyclicBarrier

忘是亡心i 2022-09-02 15:53 104阅读 0赞

## 简介 ##

CyclicBarrier是一个**同步辅助类**，允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。

注意比较[CountDownLatch][]和CyclicBarrier：  
(01) CountDownLatch的作用是允许1或N个线程等待其他线程完成执行；而CyclicBarrier则是允许N个线程相互等待。  
(02) CountDownLatch的计数器无法被重置；CyclicBarrier的计数器可以被重置后使用，因此它被称为是循环的barrier。

## CyclicBarrier类结构 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MDQ0NDE5_size_16_color_FFFFFF_t_70][]

##  CyclicBarrier类源码 ##

package java.util.concurrent;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    public class CyclicBarrier {
    
        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;
    
        private void nextGeneration() {
            trip.signalAll();
            count = parties;
            generation = new Generation();
        }
    
        private void breakBarrier() {
            generation.broken = true;
            count = parties;
            trip.signalAll();
        }
    
        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) {
                    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();
            }
        }
    
        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);
        }
    
        public int getParties() {
            return parties;
        }
    
        public int await() throws InterruptedException, BrokenBarrierException {
            try {
                return dowait(false, 0L);
            } catch (TimeoutException toe) {
                throw new Error(toe);
            }
        }
    
        public int await(long timeout, TimeUnit unit)
            throws InterruptedException,
                   BrokenBarrierException,
                   TimeoutException {
            return dowait(true, unit.toNanos(timeout));
        }
    
        public boolean isBroken() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return generation.broken;
            } finally {
                lock.unlock();
            }
        }
    
        public void reset() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                breakBarrier();
                nextGeneration();
            } finally {
                lock.unlock();
            }
        }
    
        public int getNumberWaiting() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return parties - count;
            } finally {
                lock.unlock();
            }
        }
    }

CyclicBarrier是通过[ReentrantLock][](独占锁)和[Condition][]来实现的。下面，我们分析CyclicBarrier中的核心函数: 构造函数, await()作出分析。

**1. 构造函数**

CyclicBarrier的构造函数共2个：**CyclicBarrier(int parties)** 和 **CyclicBarrier(int parties, Runnable barrierAction)**。第1个构造函数是调用第2个构造函数来实现的，直接看第2个构造函数的源码。

public CyclicBarrier(int parties, Runnable barrierAction) {
            if (parties <= 0) throw new IllegalArgumentException();
            // parties表示“必须同时到达barrier的线程个数”。
            this.parties = parties;
            // count表示“处在等待状态的线程个数”。
            this.count = parties;
            // barrierCommand表示“parties个线程到达barrier时，会执行的动作”。
            this.barrierCommand = barrierAction;
        }

**2. 等待函数await()**

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

await()方法调用的dowait方法：

private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        // 获取“独占锁(lock)”
        lock.lock();
        try {
            // 保存“当前的generation”
            final Generation g = generation;
    
            // 若“当前generation已损坏”，则抛出异常。
            if (g.broken)
                throw new BrokenBarrierException();
    
            // 如果当前线程被中断，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程。
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }
    
           // 将“count计数器”-1
           int index = --count;
           // 如果index=0，则意味着“有parties个线程到达barrier”。
           if (index == 0) {  // tripped
               boolean ranAction = false;
               try {
                   // 如果barrierCommand不为null，则执行该动作。
                   final Runnable command = barrierCommand;
                   if (command != null)
                       command.run();
                   ranAction = true;
                   // 唤醒所有等待线程，并更新generation。
                   nextGeneration();
                   return 0;
               } finally {
                   if (!ranAction)
                       breakBarrier();
               }
           }
    
            // 当前线程一直阻塞，直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，
            // 当前线程才继续执行。
            for (;;) {
                try {
                    // 如果不是“超时等待”，则调用awati()进行等待；否则，调用awaitNanos()进行等待。
                    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();
                    }
                }
    
                // 如果“当前generation已经损坏”，则抛出异常。
                if (g.broken)
                    throw new BrokenBarrierException();
    
                // 如果“generation已经换代”，则返回index。
                if (g != generation)
                    return index;
    
                // 如果是“超时等待”，并且时间已到，则通过breakBarrier()终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程。
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            // 释放“独占锁(lock)”
            lock.unlock();
        }
    }

**说明**：dowait()的作用就是**让当前线程阻塞**，**直到**“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，当前线程才继续执行。  
(01) generation是CyclicBarrier的一个成员遍历，它的定义如下：

private Generation generation = new Generation();
    
    private static class Generation {
        boolean broken = false;
    }

在CyclicBarrier中，**同一批****的线程属于****同一代****，即同一个Generation**；CyclicBarrier中通过generation对象，记录属于哪一代。  
*当有parties个线程到达barrier，generation就会被更新换代*。

(02) 如果当前线程被中断，即Thread.interrupted()为true；则通过breakBarrier()终止CyclicBarrier。breakBarrier()的源码如下：

private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }

breakBarrier()会设置当前中断标记broken为true，意味着“将该Generation中断”；同时，设置count=parties，即重新初始化count；最后，通过signalAll()唤醒CyclicBarrier上所有的等待线程。

(03) 将“count计数器”-1，即--count；然后判断是不是“有parties个线程到达barrier”，即index是不是为0。  
当index=0时，如果barrierCommand不为null，则执行该barrierCommand，barrierCommand就是我们创建CyclicBarrier时，传入的Runnable对象。然后，调用nextGeneration()进行换代工作，nextGeneration()的源码如下：

private void nextGeneration() {
        trip.signalAll();
        count = parties;
        generation = new Generation();
    }

首先，它会调用signalAll()唤醒CyclicBarrier上所有的等待线程；接着，重新初始化count；最后，更新generation的值。

(04) 在for(;;)循环中。timed是用来表示当前是不是“超时等待”线程。如果不是，则通过trip.await()进行等待；否则，调用awaitNanos()进行超时等待。

## **示例** ##

1、新建5个线程，这5个线程达到一定的条件时，它们才继续往后运行。

package cn.com.boco.HermesService;
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    public class Test {
    
    	private static int SIZE = 5;
    	private static CyclicBarrier cb;
    
    	public static void main(String[] args) {
    		cb = new CyclicBarrier(SIZE);
    		// 新建5个任务
    		for (int i = 0; i < SIZE; i++)
    			new InnerThread().start();
    	}
    
    	static class InnerThread extends Thread {
    		public void run() {
    			try {
    				System.out.println(Thread.currentThread().getName() + " wait for CyclicBarrier.");
    				// 将cb的参与者数量加1
    				cb.await();
    				// cb的参与者数量等于5时，才继续往后执行
    				System.out.println(Thread.currentThread().getName() + " continued.");
    			} catch (BrokenBarrierException e) {
    				e.printStackTrace();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MDQ0NDE5_size_16_color_FFFFFF_t_70 1][]

**结果说明**：主线程中新建了5个线程，所有的这些线程都调用cb.await()等待。所有这些线程一直等待，直到cb中所有线程都达到barrier时，这些线程才继续运行！

2、新建5个线程，当这5个线程达到一定的条件时，执行某项任务。

package cn.com.boco.HermesService;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class Test {
    
    	private static int SIZE = 5;
    	private static CyclicBarrier cb;
    
    	public static void main(String[] args) {
    		cb = new CyclicBarrier(SIZE, new Runnable() {
    			public void run() {
    				System.out.println("CyclicBarrier's parties is: " + cb.getParties());
    			}
    		});
    		// 新建5个任务
    		for (int i = 0; i < SIZE; i++)
    			new InnerThread().start();
    	}
    
    	static class InnerThread extends Thread {
    		public void run() {
    			try {
    				System.out.println(Thread.currentThread().getName() + " wait for CyclicBarrier.");
    				// 将cb的参与者数量加1
    				cb.await();
    				// cb的参与者数量等于5时，才继续往后执行
    				System.out.println(Thread.currentThread().getName() + " continued.");
    			} catch (BrokenBarrierException e) {
    				e.printStackTrace();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MDQ0NDE5_size_16_color_FFFFFF_t_70 2][]

**结果说明**：主线程中新建了5个线程，所有的这些线程都调用cb.await()等待。所有这些线程一直等待，直到cb中所有线程都达到barrier时，执行新建cb时注册的Runnable任务。

[CountDownLatch]: https://blog.csdn.net/qq_35044419/article/details/119348885
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MDQ0NDE5_size_16_color_FFFFFF_t_70]: /images/20220829/c6f7087b58e34f118c0c671a5a7c63e7.png
[ReentrantLock]: https://blog.csdn.net/qq_35044419/article/details/117021328
[Condition]: https://blog.csdn.net/qq_35044419/article/details/116981286
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MDQ0NDE5_size_16_color_FFFFFF_t_70 1]: /images/20220829/ff86fd01f47548c0af66896656bcecc8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MDQ0NDE5_size_16_color_FFFFFF_t_70 2]: /images/20220829/6c5ba5be20f3426d879c3185bd815a67.png