并发编程之CyclicBarrier原理与使用

点赞再看，养成习惯，搜一搜【一角钱技术】关注更多原创技术文章。本文 GitHub org_hejianhui/JavaStudy 已收录，有我的系列文章。

前言

控制并发流程的工具类，作用就是帮助我们程序员更容易的让线程之间合作，让线程之间相互配合来满足业务逻辑。比如让线程A等待线程B执行完毕后再执行等合作策略。

控制并发流程的工具类主要有：

类 作用 说明
Semaphore 信号量，可以通过控制“许可证”的数量，来保证线程之间的配合 线程只有拿到“许可证”后才能继续运行，相比于其它的同步器，更灵活
CyclicBarrier 线程会等待，直到足够多线程达到了事先规定的数目。一旦达到触发条件，就可以进行下一步的动作 适用于线程之间相互等待处理结果的就绪场景
Phaser 和CyclicBarrier类似，但是计数可变 Java7加入的
CountDownLatch 和CyclicBarrier类似，数量递减到0时，触发动作 不可重复使用
Exchanger 让两个线程在合适时交换对象 适用场景：当两个线程工作在同一个类的不同实例上时，用于交换数据
Condition 可以控制线程的“等待”和“唤醒” 是Object.wait() 的升级版

简介

从字面意思看，这个类的中文意思是“循环栅栏”。大概的意思就是一个可循环利用的屏障。它的作用就是会让所有线程都等待完成后才会继续下一步行动。

举个例子，就像生活中我们会约朋友到某个餐厅一起吃饭，有些朋友可能会早到，有些朋友可能会晚到，但这个餐厅规定必须等到所有人到期之后才会让我们进去。这里的朋友们就各个线程，餐厅就是CyclicBarrier。

在JUC包中为我们提供了一个同步工具类能够很好的模拟这类场景，它就是CyclicBarrier类。利用CyclicBarrier类可以实现一组线程相互等待，当所有线程都到达某个屏障点后再进行后续的操作。下图演示了这一过程。


应用场景

可用于多线程计数数据，最后合并计数结果的场景。

使用CyclicBarrier实现等待的线程都被称为参与方。参与方只需要执行cyclicBarrier.await() 就可以实现等待。由于CyclicBarrier内部维护了一个显示锁，这可以知道参与方中谁最后一个执行cyclicBarrier.await() 。当最后一个线程执行完，会使得使用相应CyclicBarrier实例的其他参与方被唤醒，而最后一个线程自身不会被暂停。其流程图如下：


public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7,() ->{
            System.out.println("****召唤神龙");
        });
        for(int i = 1;i <= 7; i++){
            int finalI = i;
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + "\t 收集到第"+ finalI +"颗龙珠");
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            },String.valueOf(i)).start();
        }
    }

源码分析

CyclicBarrier 类图



CyclicBarrier是包含了 “ReentrantLock对象lock” 和 “Condition对象trip”，它是通过独占锁实现的。

其内部主要变量和方法如下：


成员变量

//同步操作锁
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 {
  boolean broken = false;
}

可以看到 CyclicBarrier 内部是通过条件队列 trip 来对线程进行阻塞的，并且其内部维护了两个 int 型的变量 parties 和 count：

  • parties 表示每次拦截的线程数，该值在构造时进行赋值；
  • count 是内部计数器，它的初始值和 parties 相同，以后随着每次 await 方法的调用而减 1，直到减为 0 就将所有线程唤醒。

CycliBarrier 有一个静态内部类 Generation，该类的对象代表栅栏的当前代，就像玩游戏时代表的本局有些，利用它可以实现循环等待。barrierCommand 表示换代前执行的任务，当 count 减为 0 时表示本局游戏结束，需要转到下一句。在转到下一句游戏之前会将所有阻塞的线程唤醒，在唤醒所有线程之前你可以通过指定 barrierCommand 来执行自己的任务。

构造函数

主要提供了两个构造方法

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

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    // parties表示“必须同时到达barrier的线程个数”。
    this.parties = parties;
    // count表示“处在等待状态的线程个数”。
    this.count = parties;
    // barrierCommand表示“parties个线程到达barrier时，会执行的动作”。
    this.barrierCommand = barrierAction;
}

解析：

  • parties 是参与线程的个数
  • 第二个构造方法有一个Runnable参数，这个参数的意思是最后一个到达线程要执行的动作。

重要方法

CyclicBarrier类最主要的功能就是使先到达屏障点的线程阻塞并等待后面的线程，其中它提供了两种等待的方法，分别是定时等待和非定时等待。

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()表示总结已经到达栅栏
  • BrokenBarrierException表示栅栏已经被破坏，破坏的原因可能是其中一个线程await()时被中断或者超时。

dowait()方法

可以看到不管是定时等待还是非定时等待，它们都调用了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.打翻当前栅栏
      //2.唤醒拦截的所有线程
      //3.抛出中断异常
      breakBarrier();
      throw new InterruptedException();
    }
    //每次都将计数器的值减1
    int index = --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();
        }
      }
    }
 
    //如果计数器不为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();
  }
}

上面执行的代码相对比较容易看懂，我们再来看一下执行流程：


  1. 获得显示锁，判断当前线程状态是否被中断，如果是，则执行 breakBarrier 方法，唤醒之前阻塞的所有线程，并将计数器重置；
  2. 计数器 count 减 1，如果 count == 0，表示最后一个线程达到栅栏，接着执行之前指定的 Runnable 接口，同时执行 nextGeneration 方法进入下一代；
  3. 否则，进入自旋，判断当前线程是进入定时等待还是非定时等待，如果在等待过程中被中断，执行 breakBarrier 方法，唤醒之前阻塞的所有线程；
  4. 判断是否是因为执行 breakBarrier 方法而被唤醒，如果是，则抛出异常；
  5. 判断是否是正常的换代操作而被唤醒，如果是，则返回计数器的值；
  6. 判断是否是超时而被唤醒，如果是，则唤醒之前阻塞的所有线程，并抛出异常；
  7. 释放锁。

breakBarrier()方法

private void breakBarrier() {
    generation.broken = true;//栅栏被打破
    count = parties;//重置count
    trip.signalAll();//唤醒之前阻塞的线程
}

nextGeneration()方法

private void nextGeneration() {
    //唤醒所以的线程
    trip.signalAll();
    //重置计数器
    count = parties;
    //重新开始
    generation = new Generation();
}

reset()方法

接下来看看栅栏重置的方法

// 重置barrier到初始状态，所有还在等待中的线程最终会抛出BrokenBarrierException。
public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

其它方法

CyclicBarrier 其它还提供了例如getParties，isBroken，getNumberWaiting等方法，都比较简单，其中除了getParties由于parties被final修饰不可变，其余方法都会先去获得互斥锁。

/**
 * 获取当前这一轮是否已经broken。
 */
public boolean isBroken() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return generation.broken;
    } finally {
        lock.unlock();
    }
}

/**
 * 获得当前在barrier中等待的线程数。
 */
public int getNumberWaiting() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return parties - count;
    } finally {
        lock.unlock();
    }
}

总结

CountDownLatch和CyclicBarrier区别

CountDownLatch和CyclicBarrier都能够实现线程之间的等待，只不过它们侧重点不同：

  • CountDownLatch一般用于一个或多个线程，等待其他线程执行完任务后，再才执行；
  • CyclicBarrier一般用于一组线程互相等待至某个状态，然后这一组线程再同时执行；
  • CountDownLatch 是一次性的，CyclicBarrier 是可循环利用的；
  • CountDownLathch是一个计数器，线程完成一个记录一个，计数器递减，只能用一次。如下图：
  • CyclicBarrier的计数器更像一个阀门，需要所有线程都到达，然后继续执行，计数器递减，提供reset功能，可以多次使用。如下图：

PS：以上代码提交在 Github ：https://github.com/Niuh-Study/niuh-juc-final.git

文章持续更新，可以公众号搜一搜「 一角钱技术 」第一时间阅读， 本文 GitHub org_hejianhui/JavaStudy 已经收录，欢迎 Star。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 220,809评论 6赞 513
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 94,189评论 3赞 395
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 167,290评论 0赞 359
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 59,399评论 1赞 294
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 68,425评论 6赞 397
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 52,116评论 1赞 308
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,710评论 3赞 420
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,629评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 46,155评论 1赞 319
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 38,261评论 3赞 339
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,399评论 1赞 352
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 36,068评论 5赞 347
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,758评论 3赞 332
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,252评论 0赞 23
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,381评论 1赞 271
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,747评论 3赞 375
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 45,402评论 2赞 358