CyclicBarrier源码解读

傷城~ 2021-09-02 08:14 596阅读 0赞

#### CyclicBarrier ####

本意是循环栅栏，顾名思义，它可以循环使用。它是一个同步辅助器，它允许一个或多个线程阻塞等待直到其他线程执行完一系列的操作之后到达一个共同的屏障再继续执行。有点类似百米跨栏，按时你每跨一个栏必须等待其他选手也到跨栏处。  
CountDownLatch与它有点类似，但CountDownLatch只能使用一次。可以参考：[CountDownLatch源码解读][CountDownLatch]  
CyclicBarrier的实现不再是基于AbstractQueuedSynchronizer，而是采用ReentrantLock与Condition实现。

#### 举个例子 ####

import com.beust.jcommander.internal.Lists;
    import org.junit.Test;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class CyclicBarrierTest { 
    
    	private final AtomicInteger poolNumber = new AtomicInteger(1);
    
    	/** * 定义一个线程池 */
    	private ThreadPoolExecutor executor = new ThreadPoolExecutor(50, 50, 60, TimeUnit.SECONDS,
    			new LinkedBlockingDeque<>(2000), r -> new Thread(r, "工人" + poolNumber.incrementAndGet()));
    
    	class Worker implements Runnable { 
    
    		private CyclicBarrier cyclicBarrier;
    
    		Worker(CyclicBarrier cyclicBarrier) { 
    			this.cyclicBarrier = cyclicBarrier;
    		}
    
    		@Override
    		public void run() { 
    			try { 
    				Thread.sleep(new Random().nextInt(10000));
    			} catch (InterruptedException e) { 
    				e.printStackTrace();
    			}
    			System.out.println(Thread.currentThread().getName() + "执行任务结束");
    			try { 
    				cyclicBarrier.await();
    			} catch (InterruptedException | BrokenBarrierException e) { 
    				e.printStackTrace();
    			}
    		}
    	}
    
    	@Test
    	public void testCountDownLatch() { 
    		CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> System.out.println("一轮结束,执行barrierCommand"));
    		List<FutureTask> futureTaskList = Lists.newArrayList();
    		for (int i = 0; i < 10; i++) { 
    			futureTaskList.add((FutureTask) executor.submit(new Worker(cyclicBarrier)));
    		}
    		for (FutureTask futureTask : futureTaskList) { 
    			try { 
    				futureTask.get();
    			} catch (InterruptedException | ExecutionException e) { 
    				e.printStackTrace();
    			}
    		}
    	}
    }

例子中CyclicBarrier的parties定义为5，同时barrierAction为() -> System.out.println(“一轮结束,执行barrierCommand”)，表示每有5个线程通过栅栏，打印一次“一轮结束,执行barrierCommand”。

###### 执行结果 ######

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70_pic_center]

#### 源码 ####

相比CountDownLatch，CyclicBarrier主要的方法只有await，await方法既负责了信号量减小，同时唤醒下一轮线程的职责。

// 阻塞直到所有的parties执行的await方法（例子中的await为5）
        public int await() throws InterruptedException, BrokenBarrierException { 
            try { 
                return dowait(false, 0L);
            } catch (TimeoutException toe) { 
                throw new Error(toe); // cannot happen
            }
        }

// 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();
    			// 有一个线程被中断了，整个Barrier就破坏
                if (Thread.interrupted()) { 
                    breakBarrier();
                    throw new InterruptedException();
                }
    			//核心代码
    			// count减1
                int index = --count;
                if (index == 0) {   // tripped
                	// 如果count为0，执行barrierCommand，执行下轮同步循环
                    boolean ranAction = false;
                    try { 
                        final Runnable command = barrierCommand;
                        if (command != null)
                            command.run();
                        ranAction = true;
                        nextGeneration();
                        return 0;
                    } finally { 
                        if (!ranAction)
                            breakBarrier();
                    }
                }
    
                // loop until tripped, broken, interrupted, or timed out
                for (;;) { 
                    try { 
                        if (!timed)
                        	// 如果count不等于0，且不带超时，则将线程阻塞在condition条件队列trip中
                            trip.await();
                        else if (nanos > 0L)
                        	// 如果count不等于0，且带超时，则将线程定时阻塞在condition条件队列trip中
                            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();
            }
        }

当count为0时执行nextGeneration：  
1、唤醒前一轮阻塞在条件变量trip的变量让其继续执行  
2、将count恢复为parties  
3、新开一轮循环

/** * Updates state on barrier trip and wakes up everyone. * Called only while holding lock. */
        private void nextGeneration() { 
            // signal completion of last generation
            trip.signalAll();
            // set up next generation
            count = parties;
            generation = new Generation();
        }

#### 小结 ####

1、CyclicBarrier采用ReentrantLock和Condition实现了循环栅栏的功能，当count不等于0时，将线程阻塞在条件变量上，当count等于0时，通过signallAll唤醒阻塞线程并开启下一轮循环。  
2、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 */
        private final Runnable barrierCommand;

变量采用了final不可变参数保证线程安全。

[CountDownLatch]: https://blog.csdn.net/weixin_37968613/article/details/109625435
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20210518/ef99e5a69d414c548f1ba1abe7cc60ec.png