CyclicBarrier源码简析

柔情只为你懂 2023-03-13 12:15 1阅读 0赞

之前一篇文章讲了一下[CountdownLatch][]，接下来就来讲讲CyclicBarrier，两者有一些相似的地方，也有一些不同，先通过一段demo来了解一下CyclicBarrier的使用

public class CyclicBarrierTest { 
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("所有线程执行完毕");
                }
            });
    
            new Thread(){ 
                @Override
                public void run() { 
                    try { 
                        System.out.println("线程1执行逻辑");
                        cyclicBarrier.await();
                        System.out.println("线程1继续执行");
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }.start();
    
            new Thread(){ 
                @Override
                public void run() { 
                    try { 
                        System.out.println("线程2执行逻辑");
                        cyclicBarrier.await();
                        System.out.println("线程2继续执行");
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }.start();
    
            new Thread(){ 
                @Override
                public void run() { 
                    try { 
                        System.out.println("线程3执行逻辑");
                        cyclicBarrier.await();
                        System.out.println("线程3继续执行");
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }.start();
        }
    }

输出

线程1执行逻辑
    线程2执行逻辑
    线程3执行逻辑
    所有线程执行完毕
    线程3继续执行
    线程1继续执行
    线程2继续执行

demo中使用到了CyclicBarrier的地方，一共就两处，一个是CyclicBarrier的构造方法，一个是CyclicBarrier的await方法，下面来分别看看两个方法的源码

// parties：在parties个线程都到达屏障之后，这些线程才继续往下执行，可以理解为类似CountDownLatch的倒计时，后面我都用屏障倒计时来描述
    // barrierAction：触发屏障之后执行的方法，如果是null的话则不执行任何操作
    public CyclicBarrier(int parties, Runnable barrierAction) { 
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = 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 { 
        	// 每次使用屏障都会代表一个generation实例
        	// 只有在屏障被触发或是重置的时候generation才会改变
            final Generation g = generation;
    
    		// generation中只有一个broken属性，默认为false
            if (g.broken)
                throw new BrokenBarrierException();
    
            if (Thread.interrupted()) { 
            	// 打破屏障，顾名思义，运行到这里的话，所有被阻塞的线程都会继续往下执行
                breakBarrier();
                throw new InterruptedException();
            }
    
    		// 在构造方法中，count的值等于parties，即屏障倒计时，在这里倒计时减1
            int index = --count;
            // 当屏障倒计时为0，即所有的线程都到达屏障之后
            if (index == 0) {   // tripped
                boolean ranAction = false;
                try { 
                    final Runnable command = barrierCommand;
                    if (command != null)
                    	// 运行我们要执行的CyclicBarrier构造方法中传入的Runnable方法
                        command.run();
                    ranAction = true;
                   	// 这个方法中，也会让所有被阻塞的线程继续往下执行
                   	// 前面我们提到了generation在屏障被触发时会改变，所以这里new了一个新的generation实例，已经跟上面的实例g不一样了
                    nextGeneration();
                    return 0;
                } finally { 
                	// 如果ranAction这里是false，那么说明我们在执行CyclicBarrier构造方法中传入的Runnable方法时报错了
                    if (!ranAction)
                        breakBarrier();
                }
            }
    
            // loop until tripped, broken, interrupted, or timed out
            for (;;) { 
                try { 
                	// timed是我们传入的参数，是false
                    if (!timed)
                    	// private final Condition trip = lock.newCondition();
                    	// 这里的trip就是我们最开始加的锁的一个Condition
                    	// 在这里的作用就是在上面屏障倒计时不为0的时候阻塞住当前线程，最后实现的效果看起来就像所有线程都在屏障前等待
                    	// 通过这里的Condition，我们就能知道上面的breakBarrier和nextGeneration方法中其实就是调用了Condition的signalAll方法来唤醒被阻塞的线程
                        trip.await();
                    else if (nanos > 0L)
                    	// 传入的nanos大于0的话，则阻塞指定的时间
                        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();
    
    			// 如果之前执行过了nextGeneration方法，这里的g和generation就是不相等的，此时循环结束，方法返回
                if (g != generation)
                    return index;
    
                if (timed && nanos <= 0L) { 
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally { 
       		// 解锁
            lock.unlock();
        }
    }

总结一下CyclicBarrier方法里面的主要逻辑，首先构造方法指定了屏障倒计时和我们在CyclicBarrier配置触发后要执行的方法，await方法里面如果屏障倒计时不为0的话，那么线程就会被Condition阻塞。如果程序在运行中出现了异常，那么会执行breakBarrier方法来唤醒所有阻塞的线程；如果最后所有的线程都到达屏障之后，一切顺利，就会执行nextGeneration方法唤醒所有阻塞的线程

源码分析的差不多了，我们来类比一下CountdownLatch，我们可以发现在CountdownLatch的例子中，被阻塞的线程是调用了CountdownLatch的await方法的线程，那些调用countDown方法的线程在执行完countDown方法之后就继续执行了，只有一个线程被阻塞，而其他的线程都继续执行

反观CyclicBarrier，所有调用CyclicBarrier的await方法的线程都被阻塞了，最后当指定数量的线程都调用了await方法之后，大家被一起唤醒，然后继续往下执行

这么说可能有点不太好理解，举个例子吧，CyclicBarrier和CountdownLatch就像幼儿园的小朋友上学，老师代表主线程，小朋友代表多个工作线程。CyclicBarrier就是上学，小朋友们一个个来到幼儿园，如果有人没到，大家都得等着，只有当所有的小朋友都到了之后，老师才能开始给孩子们上课；而CountdownLatch就像放学，当小朋友的家长来了之后，就会把自家的小朋友接走，当最后一个小朋友被接走之后，老师才能走

CountdownLatch和CyclicBarrier下一个不同点就是CyclicBarrier的计数可以复用，我把nextGeneration的源码贴出来

private void nextGeneration() { 
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        // 关键在这里，当屏障被触发之后，本来count已经被递减为了0，但是这里又重置为了parties，所以又能继续使用屏障了
        count = parties;
        generation = new Generation();
    }

CountdownLatch的计数器倒计时结束之后就不能继续使用了，而CyclicBarrier再使用完之后，通过nextGeneration和breakBarrier方法又会将计数重置，达到循环利用的效果

[CountdownLatch]: https://blog.csdn.net/LO_YUN/article/details/105949051