同步工具类CyclicBarrier

痛定思痛。 2022-09-10 07:18 139阅读 0赞

使用场景：多个线程都到达某个状态后，一起进入下个阶段  
使用方法：

CyclicBarrier barrier = new CyclicBarrier(5); 
    barrier.await();

举例

public class CyclicBarrierTest {
        
        public static void main(String[] args) {
        
            CyclicBarrier barrier = new CyclicBarrier(5,()-> System.out.println("阶段结束"));
            Random random = new Random();
            for (int i = 0; i < 5; i++) {
        
                //5个人去公司面试
                new Thread(() -> {
        
                    try {
        
                        Thread.sleep(random.nextInt(5000));
                        System.out.println(Thread.currentThread().getName() + " - 到达公司");
                        barrier.await();//等待其他人到达公司
                        Thread.sleep(random.nextInt(5000));
                        System.out.println(Thread.currentThread().getName() + " - 笔试结束");
                        barrier.await();//等待其他人笔试结束
                        Thread.sleep(random.nextInt(5000));
                        System.out.println(Thread.currentThread().getName() + " - 面试结束");
                    } catch (InterruptedException | BrokenBarrierException e) {
        
                        e.printStackTrace();
                    }
                }, "线程-" + (i + 1)).start();
            }
        }
    }

![在这里插入图片描述][c3652b2407644c2eba102a8cf1faa721.png]

CyclicBarrier使用ReentrantLock+Condition实现

public class CyclicBarrier {
        
        private final ReentrantLock lock = new ReentrantLock();
        //用于线程间唤醒，最后一个结束的线程要唤醒其他线程，然后一起进入下一个阶段
        private final Condition trip = lock.newCondition();
        //线程总数，创建对象传过来的参数
        private final int parties;
        private final Runnable barrierCommand;
        //每个阶段都会创建一个generation对象，代表这个阶段的状态
        private Generation generation = new Generation();
        //还没有完成当前阶段任务的还有多少线程
        private int count;
        //...
        public CyclicBarrier(int parties, Runnable barrierAction) {
        
            if (parties <= 0) throw new IllegalArgumentException();
            //线程总数
            this.parties = parties;
            this.count = parties;
            //当所有线程被唤醒时，执行barrierCommand表示的Runnable。
            this.barrierCommand = barrierAction;
        }
    }

await方法实现

public int await() throws InterruptedException, BrokenBarrierException {
        
            try {
        
            	//false一直等待，true设置等待的时间
                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;
    			//broken抛异常
                if (g.broken)
                    throw new BrokenBarrierException();
    			//被中断了
                if (Thread.interrupted()) {
        
                	//唤醒所有线程
                    breakBarrier();
                    throw new InterruptedException();
                }
    			//count记录了没有完成任务的线程的数量，每一个线程await需要-1
                int index = --count;
                //count=0的时候，唤醒其他线程
                if (index == 0) {
          // tripped
                    boolean ranAction = false;
                    try {
        
                    	//执行配置的barrierCommand runnable
                        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)
                            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();
                        }
                    }
                    //以下是被唤醒后
    				//broken抛异常，是不是用breakBarrier的方法向下运行的
                    if (g.broken)
                        throw new BrokenBarrierException();
    				//其他线程执行到index == 0的时候会用nextGeneration设置generation = new Generation()
                    if (g != generation)
                    	//返回自己await的时候，没有完成任务的数量
                        return index;
    				//错的参数抛异常
                    if (timed && nanos <= 0L) {
        
                        breakBarrier();
                        throw new TimeoutException();
                    }
                }
            } finally {
        
                lock.unlock();
            }
        }

breakBarrier：人没到齐想执行下一个阶段可以用breakBarrier

private void breakBarrier() {
        
        	//设置generation被broken
            generation.broken = true;
            //重置未执行完的线程
            count = parties;
            //唤醒所有线程
            trip.signalAll();
        }

[c3652b2407644c2eba102a8cf1faa721.png]: /images/20220829/080e585387df4f5a8d54339960e59bd9.png