CyclicBarrier源码解析

偏执的太偏执、 2022-05-08 09:20 200阅读 0赞

`CyclicBarrier` 基于可重入锁 `ReentrantLock` 和 `Condition` 共同实现。

*  `parties`：设置多少个线程或者任务等待至 `barrier` 状态
 *  `barrierAction`：设置指定线程全部到达 `barrier` 状态时会执行的任务

public CyclicBarrier(int parties) { 
    }
    
    public CyclicBarrier(int parties, Runnable barrierAction) { 
    }

分析 `await` 方法

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

private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException { 
        final ReentrantLock lock = this.lock;
        lock.lock(); // 获取锁
        try { 
            final Generation g = generation;
            // 当前generation已损坏，则抛出异常
            if (g.broken) 
                throw new BrokenBarrierException();
            // 如果当前线程被中断，则通过breakBarrier终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程
            if (Thread.interrupted()) { 
                breakBarrier();
                throw new InterruptedException();
            }
    
            int index = --count; // 计数值自减
            if (index == 0) {   // 计数值为0表示已经有parties个线程到达barrier
                boolean ranAction = false;
                try { 
                    final Runnable command = barrierCommand; // 第二个构造方法中传递的到达barrier需要执行的任务
                    if (command != null)
                        command.run(); // 执行到达barrier的任务
                    ranAction = true;
                    nextGeneration(); // 唤醒所有等待线程并更新generation
                    return 0;
                } finally { 
                    if (!ranAction)
                        breakBarrier();
                }
            }
    
            // loop until tripped, broken, interrupted, or timed out
            // 线程处于阻塞状态，直到 parties个线程到达barrier 或 当前线程被中断 或 超时，线程才继续执行
            for (;;) { 
                try { 
                    // 没有设置超时时间，使用await等待；否则使用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 { 
                        // 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(); // 释放锁
        } 
    }