首页 > 软件编程 > java > CyclicBarrier循环栅栏

Java中的CyclicBarrier循环栅栏深入解析

2023-12-25 11:27:44 作者：Wayfreem

这篇文章主要介绍了Java中的CyclicBarrier循环栅栏深入解析,CycleBarrier 它就相当于是一个栅栏,所有线程在到达栅栏后都需要等待其他线程,等所有线程都到达后,再一起通过,需要的朋友可以参考下

关于 CyclicBarrier

Java JUC 包中提供类似的工具，可以设置一个简单的集合点，等所有成员到齐了之后，再执行下一步操作。CycleBarrier 它就相当于是一个栅栏，所有线程在到达栅栏后都需要等待其他线程，等所有线程都到达后，再一起通过。

CyclicBarrier 可以有不止一个栅栏，因为它的栅栏（Barrier）可以重复使用（Cyclic）。

CycleBarrier 特别适用于并行迭代计算，每个线程负责一部分计算，然后在栅栏处等待其他线程完成，所有线程到齐后，交换数据和计算结果，再进行下一次迭代。

初步理解

CyclicBarrier 有一个 parties 成员变量，表示参与的线程的个数，下面来看一个例子，我们通过这个例子来理解下 Cycle 的含义：

public class Demo1 {
    public static void main(String[] args) {
        // 初始化一个 CyclicBarrier，大小为 5，当 5 个线程满足条件之后，就执行一次：执行完毕！
        CyclicBarrier cyclicBarrier = new CyclicBarrier( 5, () -> {
            System.out.println("执行完毕！");
        });
        // 为了测试是否能循环，这里循环设置为 20，预期的结果应该是会执行 4 次：执行完毕！
        for (int i=0; i<20; i++) {
            int temp = i;
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + ", 执行到：" + temp );
                try{
                    cyclicBarrier.await();  // 等待其他线程到齐，到齐了之后就会执行一次
                } catch (BrokenBarrierException | InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

程序输出

由于多线程输出，我是多刷了几遍（也不知道刷了多少遍了）就会出现下面的执行结果，这里就可以看出来效果了，这里有 20个线程，5 个线程为一组（如果是按照源码上面来说，就是 5 个线程 parties 为一代 generation ），可以满足我们的预期。

每次参与的线程数为 5，然后当 5个线程都执行完毕之后，再进入下一代的计算，看到这里应该还是比较容易理解的。

源码探究

成员变量

首先，我们看下 CycleBarrier 的成员变量，

//同步操作锁
private final ReentrantLock lock = new ReentrantLock();
//线程拦截器
private final Condition trip = lock.newCondition();
//每次拦截的线程数
private final int parties;
//换代前执行的任务
private final Runnable barrierCommand;
//表示栅栏的当前代
private Generation generation = new Generation();
//计数器
private int count;
//静态内部类Generation
private static class Generation {
    Generation() {}                 // 防止访问构造函数创建
    boolean broken;                 // 初始化为 false
}

这里说明下：

trip 是一个 condition，就是用来阻塞与唤醒线程使用的。
两个 int 类型的变量 parties 与 count，在初始化 CycleBarrier 的时候，两个值的大小是一样的，之后随着线程的 await 方法的调用而 -1，直到减到 0 ，就将所有的线程全部唤醒。
内部静态变量 Generation，表示当前栅栏的代，上面的例子中，5个线程就为一代，执行完了，就进行下一代。
barrierCommond，表示换代前执行的任务，当 count 减为 0 时表示本次结束。在转到下一局游戏之前，将所有的线程唤醒，在唤醒之前可以通过指定的 barrierCommond 来执行自己的任务。

构造器

接着，我们看下构造器相关的源码，这里就可以看到 count 与 parties 的初始化的时候就是一致的

//构造器，这里就是我们上面那个例子使用的构造器
public CyclicBarrier(int parties, Runnable barrierAction) {
  if (parties <= 0) throw new IllegalArgumentException();
  this.parties = parties;
  this.count = parties;
  this.barrierCommand = barrierAction;
}
// 构造器，只传入一个 parties 的大小，不进行自定义任务
public CyclicBarrier(int parties) {
  this(parties, null);
}

await() 方法

这里我们看下 await() 方法，在上面的例子中，是使用 await() 方法等待其他线程到齐，再去执行后面的操作

// 非定时等待
public int await() throws InterruptedException, BrokenBarrierException {
  try {
    return dowait(false, 0L);
  } catch (TimeoutException toe) {
    throw new Error(toe);
  }
}
// 定时等待
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException {
  return dowait(true, unit.toNanos(timeout));
}

看源码之后，发现 await() 方法最后都是调用 dowait() 方法，我们就直接看下核心的 dowait 方法的执行逻辑，会发现这块并没有特别晦涩难懂，那么我们就一起继续向下读

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()) {
            // 如果中断了，就会执行下面三件事
            // 1. 将 generation 的 broken 标记设置为 true
            // 2. 重置 count 值
            // 3. 唤醒当前拦截的线程
            breakBarrier();
            throw new InterruptedException();
        }
        int index = --count;    // 每次都将计数器的值减 1
        if (index == 0) {  // // 计数器的值减为0，就会唤醒所有线程并转换到下一代
            Runnable command = barrierCommand;
            if (command != null) {    // // 如果存在自定义的任务，这里就会做判断，在唤醒所有线程之前执行
                try {
                    command.run();
                } catch (Throwable ex) {
                    breakBarrier();
                    throw ex;
                }
            }
            nextGeneration();    // 唤醒所有线程，并转到下一代
            return 0;
        }
        //官方注释： loop until tripped, broken, interrupted, or timed out
        //这里循环是：如果计数器不为0则执行此循环
        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 {
                     //若在捕获中断异常前已经完成在栅栏上的等待, 则直接调用中断操作
                    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();    // 解锁
    }
}

nextGeneration() 与 breakBarrier()

后面，我们看下刚刚上面提到过的切换代 nextGeneration()，打破栅栏方法 breakBarrier()

//切换栅栏到下一代
private void nextGeneration() {
  //唤醒条件队列所有线程
  trip.signalAll();
  //设置计数器的值为需要拦截的线程数
  count = parties;
  //重新设置栅栏代次
  generation = new Generation();
}
//打破当前栅栏
private void breakBarrier() {
  //将当前栅栏状态设置为打翻
  generation.broken = true;
  //设置计数器的值为需要拦截的线程数
  count = parties;
  //唤醒所有线程
  trip.signalAll();
}

reset()

最后，我们看下怎么重置，这里其实就是调用了上面说的切换代 nextGeneration() 与 breakBarrier() 方法

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

回顾

最后看一个例子回顾下

public class Demo {
    static class Tourist extends Thread {
        CyclicBarrier cyclicBarrier;
        public Tourist(CyclicBarrier cyclicBarrier){
            this.cyclicBarrier = cyclicBarrier;
        }
        @Override
        public void run() {
            try {
                // 模拟点先各自独立运行
                Thread.sleep( (int) (Math.random() * 1000) );
                // 集合点 A
                cyclicBarrier.await();
                System.out.println(this.getName() + "，到达集合点 A " + System.currentTimeMillis() );
                // 集合后模拟再各自独立运行
                Thread.sleep( (int) (Math.random() * 1000));
                // 集合点 B
                cyclicBarrier.await();
                System.out.println(this.getName() + "，到达集合点 B " + System.currentTimeMillis() );
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        // 定义出多个 线程
        int num = 3;
        Tourist[] tourist = new Tourist[num];
        CyclicBarrier barrier = new CyclicBarrier(3, () -> {
            System.out.println("集合完毕： " + System.currentTimeMillis() + " 执行的线程为： " + Thread.currentThread().getName());
        });
        for (int i = 0; i< num; i++){
            tourist[i] = new Tourist(barrier);
            tourist[i].start();
        }
    }
}

到此这篇关于Spring中的CyclicBarrier循环栅栏深入解析的文章就介绍到这了,更多相关CyclicBarrier循环栅栏内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！