Java并发之CyclicBarrier

技术
,

CyclicBarrier是juc包下一个并发辅助类，类似于CountDownLatch，但又不同，CyclicBarrier保证一组线程在同一个地方互相等待，直到最后一个线程到来后，然后一起再继续向下执行。

CyclicBarrier提供了一个叫做Generation(代/批)的概念，每个Generation的线程由该Generation最后一个到达的唤醒其他先到的线程，一个CyclicBarrier允许有N个Generation。

CyclicBarrier初始化支持两个参数:

  • parties，int类型，标记多少个线程等待后，全部唤醒继续执行
  • barrierAction，Runnable类型，如果不为null，最后一个到达barrier的线程会执行此Runnable

最主要方法await，内部主要依赖Lock、Condition以及一个count计数来工作，count初始值为构造参数parties的值

  • Lock保证线程同步，同一时刻只有一个线程能够执行await的核心逻辑
  • 核心逻辑先做--count操作，然后判断count计数的值是否为0
    • 如果不是，在Condition下等待
    • 如果是，执行barrierAction，然后唤醒所有等待在Condition的线程，最后新初始化一个Generation，并将parties的值赋值给count(重置计数)

注意：如果parties=2，不要同时启动4个线程，因为并不能完全保证不同Generation的线程是顺序唤醒的，因为内部公用同一把Lock

代码示例：模拟轮渡载满才过河

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public class Ferry {

  public static void main(String[] args) throws Throwable {
    int count = 4;
    CyclicBarrier cyclicBarrier = new CyclicBarrier(count, new Runnable() {

      @Override
      public void run() {
        System.out.println("船开了");
      }
    });
    Thread[] threads = new Thread[count];
    for (int i = 0; i < count; i++) {
      Thread t = new Thread(new Runnable() {

        @Override
        public void run() {
          try {
            System.out.println(Thread.currentThread().getName() + "等待过河");
            cyclicBarrier.await();
            System.out.println(Thread.currentThread().getName() + "过河成功");
          } catch (InterruptedException e) {
          } catch (BrokenBarrierException e) {
          }
        }
      });
      threads[i] = t;
    }
    for (int i = 0; i < count; i++) {
      threads[i].start();
    }

    Thread.sleep(10000);
  }
}

输出如下：

1
2
3
4
5
6
7
8
9
Thread-3等待过河
Thread-2等待过河
Thread-0等待过河
Thread-1等待过河
船开了
Thread-1过河成功
Thread-0过河成功
Thread-2过河成功
Thread-3过河成功

可以发现最后到达的线程会先执行(这个毋庸置疑)，其他线程随机顺序，这主要是因为内部的Lock使用的是非公平的，所以并非先到先执行。

CyclicBarrier可以保证多个线程在某一处互相等待，然后继续执行，在某些特定的场景下使用会非常方便、有效。