jdk源码解析四之CyclicBarrier

逃离我推掉我的手 2023-02-20 12:08 67阅读 0赞

### 文章目录 ###

*  CyclicBarrier
 *   *  await
     *  reset

# CyclicBarrier #

阻塞一组线程直到某件事发生,可重复使用,另一种栅栏式。5个人约好集合后去其他地方。

public CyclicBarrier(int parties, Runnable barrierAction) {
        
            if (parties <= 0) throw new IllegalArgumentException();
            //屏障拦截的线程数量
            this.parties = parties;
            //线程数量
            this.count = parties;
            //指定个数线程被wait后,执行的后续方法
            this.barrierCommand = barrierAction;
        }

## await ##

public int await() throws InterruptedException, BrokenBarrierException {
        
            try {
        
                return dowait(false, 0L);
            } catch (TimeoutException toe) {
        
                throw new Error(toe); // cannot happen
            }
        }
    
    ```java
    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;
                        //重新构造可wait线程数
                        //以及换代
                        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 {
        
                         //说明这个线程不是当代的,就不会影响当代执行,则中断线程
                            //如果在等待过程中，线程被中断了，就抛出异常。但如果中断的线程所对应的CyclicBarrier不是这代的，
                            // 比如，在最后一次线程执行signalAll后，并且更新了这个“代”对象。在这个区间，
                            // 这个线程被中断了，那么，JDK认为任务已经完成了，就不必在乎中断了，只需要打个标记。
                            // 该部分源码已在dowait(boolean, long)方法中进行了注释
                            Thread.currentThread().interrupt();
                        }
                    }
    
                    if (g.broken)
                        throw new BrokenBarrierException();
    
                    //表示正常换代了,返回当前线程所在下标,便于下次做选举等操作
                    //仔细发现,栅栏不是一次性的,可以重复使用多次,但是锁是同一个,所以需要用generation进行区分,是否是同一个栅栏
                    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个数
            count = parties;
            //换代
            generation = new Generation();
        }
    
        private void breakBarrier() {
        
            //标记栅栏损坏
            generation.broken = true;
            //count恢复parties个数
            count = parties;
            //唤醒所有线程
            trip.signalAll();
        }

## reset ##

public void reset() {
        
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
        
                breakBarrier();   // break the current generation
                nextGeneration(); // start a new generation
            } finally {
        
                lock.unlock();
            }
        }

总结:可重复使用多次,底层采用显示锁进行加锁,  
索引按照初始count递减,当索引为0时,执行栅栏任务  
当执行成功,通知当代所有线程,接着换代,如果当代有线程中断了,且已经执行了换代任务了,这个时候会将中断线程标记中断