JUC之CyclicBarrier

雨点打透心脏的1/2处 2022-10-14 00:54 122阅读 0赞

在《[JUC之CountDownLatch][JUC_CountDownLatch]》中介绍了**CountDownLatch**实现闭锁的功能，在JDK中还提供了另外一个具有相同功能的类-**CyclicBarrier**。通过学习CyclicBarrier不仅要了解CyclicBarrier的实现原理，还需要了解其与CountDownLatch的不同之处。

![20210603194306462.png][]

CyclicBarrier的实现没有直接使用内部类来继承**AbstractQueuedSynchronizer**，而是使用**ReentrantLock**与**ConditionObject**来实现功能，这也可以看做是间接依赖了AbstractQueuedSynchronizer。

--------------------

CyclicBarrier，拆分单词Cyclic与Barrier。

Cyclic百度翻译：循环的

![20210603194401996.png][]

Barrier百度翻译：屏障

![20210603194432228.png][]

CyclicBarrier经过翻译之后，即循环的屏障。所以CyclicBarrier具有两个特点：循环的、屏障。

接下来，我们从这两个特点出发，了解CyclicBarrier的实现原理和功能。

--------------------

# **█ 构造器** #

CyclicBarrier具有两个构造器。

（1）一个int类型的参数，用于指定屏障的数量。

public CyclicBarrier(int parties) {
        this(parties, null);
    }

（2）重载构造器，一个int类型的参数，一个**Runnable**类型的参数。int参数指定屏障的数量，Runnable参数指定当屏障被突破之后就执行Runnable的**run**方法，即突破屏障之后的后续动作（此处在**await**方法的代码中有体现）。

public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        // 初始化屏障的数量
        this.parties = parties;
        // 用于记录线程的访问数量
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

--------------------

# **█ 关于屏障** #

在CountDownLatch中也有屏障，也可以设定屏障的数量。设定了屏障的数量之后，每个线程调用countDown去减少屏障的数量，当屏障的数量减少到0以后，await方法便不再阻塞等待。在屏障的数量减少到0之前，await方法会一直阻塞等待（CountDownLatch看做是做减法的屏障）。

CyclicBarrier中的屏障与CountDownLatch不同，CountDownLatch是对屏障的数量做减法，而CyclicBarrier是对屏障的数量做加法。即设定好屏障的数量之后，await方法会一直阻塞等待，只有当线程的数量达到了屏障的数量之后，await才不会阻塞等待。（每有一个线程调用await，计数器就会加1，当计数器的数量达到了屏障数量之后，await才会结束阻塞等待，继续往下运行）。

这里可以做个比喻，比如乘坐公交车，规定乘坐人数必须达到50人才能发车（设置屏障的数量）。只有公交车的人数达到了50人，公交车才会发车（await方法调用次数达到了屏障树）。在人数未达到50人之前，公交车是始终不动的（阻塞等待）。

CyclicBarrier中维护了一个int类型的字段来记录屏障的数量：

private final int parties;

从功能上来看的话，CyclicBarrier可以看做是一个做加法的屏障，但其实现方式其实在对一个数做减法，这个数初始值为构造函数设置的屏障数量，当一个线程访问了await，就将该数字减1。当数字等于0时，表示达到了设置的屏障数量，便可以停止阻塞等待，方法继续运行了。在内部维护了一个int类型的变量**count**来表示这个数。

private int count;

在构造函数中初始化：

public CyclicBarrier(int parties, Runnable barrierAction) {
        ......
        this.count = parties;
        ......
    }

--------------------

# **█ **await #

*  **await**

await方法便是整个CyclicBarrier的功能入口了。

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 {
        // 使用ReentrantLock独占锁，保证每次同时只有一个线程能够执行该方法逻辑           
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 因为CyclicBarrier是可循环使用的嘛，用Generation来标记循环
            final Generation g = generation;
            // broken为true，表示当前屏障已经被突破了
            if (g.broken)
                throw new BrokenBarrierException();
    
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }
            // 看这里，是对数量做减法，一次线程访问就减1
            int index = --count;
            // 当数量减少到0，即表示可以不用阻塞等待了
            if (index == 0) { 
                boolean ranAction = false;
                try {
                    // 如果构造器中指定了Runnable，则调用Runnable的run方法。
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
    
            // 线程访问数没有达到屏障的数量，当前线程会一直阻塞等待
            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();
                    }
                }
    
                if (g.broken)
                    throw new BrokenBarrierException();
    
                if (g != generation)
                    return index;
    
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

①使用ReentrantLock独占锁，保证在并发条件下，每次同时只能有一个线程获取到锁，继续往下执行方法的逻辑。

②--count，这里可以看出，其实CyclicBarrier是通过做减法来实现的。count初始值为构造函数中指定的屏障数量，一个线程访问了之后，数量就会减1。（感觉其实也可以做减法，count初始值为0，一个线程访问了，数量就加1，直到值与屏障parties的数量相等）

③index == 0，对count做减法，当数量变成0之后，表示线程数量已经达到了屏障树。

if (index == 0) { 
        boolean ranAction = false;
        try {
            final Runnable command = barrierCommand;
            if (command != null)
                command.run();
            ranAction = true;
            nextGeneration();
            return 0;
        } finally {
            if (!ranAction)
                breakBarrier();
        }
    }

上面的代码表示当线程的数量达到了设置的屏障数的处理逻辑。如果在构造器中指定了Runnable对象，则调用它的run方法。

**nextGeneration**，开启新一轮的屏障。CyclicBarrier的循环特点主要就体现在了这个方法里。关于nextGeneration此方法将在后面介绍。

④自旋，如果count的数量不为0，表明线程访问数没有达到设定的屏障的数量，则进入自旋的逻辑。timed是标记是否超时等待，如果为true，则达到了设定的等待时间，线程会被唤醒。

⑤线程调用了Condition的await后会释放锁，并进入等待队列。因为释放了锁，因此ReentrantLock\#lock方法会被新的线程获取。

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();
            }
        }
    
        if (g.broken)
            throw new BrokenBarrierException();
    
        if (g != generation)
            return index;
    
        if (timed && nanos <= 0L) {
            breakBarrier();
            throw new TimeoutException();
        }
    }

自旋的主要逻辑，在于调用了trip.**await**()，trip为：

private final Condition trip = lock.newCondition();

借助了AbstractQueuedSynchronizer中的ConditionObject功能，可以实现线程的等待与唤醒。调用了**await**方法，线程会一直阻塞等待，直到其他线程调用了**signal**或**signalAll**将其从等待队列中唤醒。被唤醒的线程会继续执行后面的代码。那什么时候线程会被唤醒呢，还是在nextGeneration方法中，有调用signalAll来唤醒等待队列中的所有线程。

*  **nextGeneration**

私有方法。调用Condition的signalAll方法唤醒所有的等待线程。并重置新一轮的屏障。CyclicBarrier的循环功能主要就体现在这个方法中了。

private void nextGeneration() {
        trip.signalAll();
        count = parties;
        generation = new Generation();
    }

*  **breakBarrier**

私有方法。除了等访问线程的数量达到了设定的屏障数之后，屏障被突破，所有线程停止阻塞等待，继续向下执行外，CyclicBarrier还提供了方法用于干预这种行为，在线程数没有达到屏障数之前，可以调用breakBarrier方法来手动突破屏障。在调用breakBarrier之后，await方法会抛出**BrokenBarrierException**异常。

private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }

*  **reset**

reset方法是nextGeneration、breakBarrier方法功能的集合，即突破当前屏障，并开启下一轮屏障。因为nextGeneration、breakBarrier两个方法都是private，而reset是public。因此只能通过对象调用到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();
        }
    }

--------------------

### 使用 ###

（1）使用CyclicBarrier\#await设置屏障

public static void main(String[] args) {
        int threadNum = 10;
        // 屏障数量设置成threadNum
        CyclicBarrier cyclicBarrier = new CyclicBarrier(threadNum);
    
        for (int i = 0; i < threadNum; i++) {
            new Thread(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+" before here "+System.currentTimeMillis());
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName()+" after here "+System.currentTimeMillis());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, String.valueOf(i)).start();
        }
    }

从下面的运行结果可以看出，所有的线程都是先打印“before here”，然后在线程访问数没有达到设置的屏障数10之前，所有的线程都在await处等待住。等线程数量达到了设置的屏障数10时，突破屏障，接着打印“after here”。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

（2）当线程访问数量没有达到屏障数量之前，所有线程都会阻塞等待。

public static void main(String[] args) {
        int threadNum = 10;
        // 屏障数量设置成threadNum
        CyclicBarrier cyclicBarrier = new CyclicBarrier(threadNum);
    
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+" before here");
                cyclicBarrier.await();
                System.out.println(Thread.currentThread().getName()+" after here");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "A").start();
    
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+" before here");
                cyclicBarrier.await();
                System.out.println(Thread.currentThread().getName()+" after here");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "B").start();
    }

设置的屏障数为10，但只有两个线程访问了。此时两个线程会一直阻塞。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

（3）reset重置屏障

public static void main(String[] args) {
        int threadNum = 10;
        // 屏障数量设置成threadNum
        CyclicBarrier cyclicBarrier = new CyclicBarrier(threadNum);
    
        new Thread(()->{
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "A").start();
    
        new Thread(()->{
            try {
                cyclicBarrier.reset();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "B").start();
    
        new Thread(()->{
            try {
                cyclicBarrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "C").start();
    
    }

在线程数没有达到设置的屏障数之前，调用了reset方法，await方法会抛出java.util.concurrent.BrokenBarrierException异常。

![2021060319551696.png][]

（4）循环屏障

public static void main(String[] args) {
        int threadNum = 10;
        // 屏障数量设置成threadNum
        CyclicBarrier cyclicBarrier = new CyclicBarrier(threadNum);
    
        for (int i = 0; i < threadNum * 3; i++) {
            new Thread(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+" before here "+System.currentTimeMillis());
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName()+" after here "+System.currentTimeMillis());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, String.valueOf(i)).start();
        }
    }

虽然设置的屏障数是10，但当30个线程访问的时候，也是可以达到屏障的效果的。每10个线程为1组，当一组的线程数达到屏障数之后，突破屏障，并开启新一轮的屏障。但如果访问线程的数量不是屏障的整数倍，就会存在某一轮的屏障没有达到指定数量而导致那一轮所在的所有线程都阻塞等待。

[JUC_CountDownLatch]: https://blog.csdn.net/weixin_50518271/article/details/117301908?spm=1001.2014.3001.5501
[20210603194306462.png]: /images/20221014/6ac8093abfa34177b404b25edf25c084.png
[20210603194401996.png]: /images/20221014/9eec68b5483e4804b5166bb2dfd42980.png
[20210603194432228.png]: /images/20221014/c55189b1a448415a88ef5d7b7e91b50a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20221014/a2100fc555834084a1d0f1d7029debd1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20221014/2365d5bcf9e1409b9b103a14ada7a836.png
[2021060319551696.png]: /images/20221014/96ae1b3b27f647da898b29833fb0bbd4.png