Java多线程编程-栅栏CyclicBarrier实例

news2025/1/23 12:09:56

前言

本文是基于《Java多线程编程实战指南-核心篇》第五章个人理解，源码是摘抄作者的源码，源码会加上自己的理解。读书笔记目前笔者正在更新如下，《Java多线程编程实战指南-核心篇》，《How Tomcat Works》，再到《spring 源码》解读。

栅栏CyclicBarrier

有时候多个线程可能需要相互等待对方执行到代码中某个地方（集合点），这这些线程才能继续执行。这种等待类似于大家相约去爬山的情形：大家实现约定好时间到集合点，先到的人必须在集合点等待其他未到的人，只有所有参与人员到齐之后大家才能够触发去登山。Java.util.concurrent.CyclicBarrier，该类就可以用来实现这种等待。

使用CyclicBarrier实现等待的线程被成为参与方，参与方只需要执行CyclicBarrier.await()就可以实现等待。尽管从应用代码的角度来看，参与方是并发执行CyclicBarrier.await()的，但是CyclicBarrier内部维护了一个显示锁，总是能区分出最后一个执行CyclicBarrier.await()的线程。除了最后一个线程外，任何参与方都被暂停，处于WAITING状态。最后一个线程执行await时候，会唤醒其他参与方，而最后一个线程自身不会暂停。

与CountDownLatch不同（有关CountDownLatch可以参考之前博客），在所有参与方唤醒时候，任何线程再次执行await又会暂停，直到这些线程最后一个线程执行了await

实例

实例模拟了士兵参与打靶训练，所有参与训练士兵被分为若干组Rank，每组被称为一排，一排士兵个数等于靶子的个数，每次只能够有一排士兵进行射击。一排士兵必须同时射击，射击完毕的士兵必须等待同排其他士兵射击完毕后才能整排撤离射击点。然后一排一排的轮流射击。

代码基本讲解

1.主函数ShootPractice实例化，输入参数为N（每排士兵个数=靶子个数），linecount（排数），lasting（最大持续时间），并且根据士兵数量（N*linecount）实例化了所有士兵存于rank中.

2.实例化了两个CycliBarrier，shiftBarrier和startBarrier，startBarrier保证同一时间士兵能在同一时间射击，可以理解为上图所示，士兵是否就绪，如图士兵一和四是站在射击地点了，处于就绪状态，需要等待士兵二和三。

而shiftBarrier，是确保所有士兵射击后同时撤离射击位置，这是由于有的士兵有的已经完成射击，有的还没有，如图士兵一和二已经射击结束，三和四还在射击，所以士兵一和二需要等待三和四。

3.所以在线程run函数的，在开火fire函数前后有两个等待，一个是等待所有人就绪，一个是等待所有人射击完成。

代码

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class ShootPractice {
  // 参与打靶训练的全部士兵
  final Soldier[][] rank;
  // 靶的个数，即每排中士兵的个数
  final int N;
  // 打靶持续时间（单位：秒）
  final int lasting;
  // 标识是否继续打靶
  volatile boolean done = false;
  // 用来指示进行下一轮打靶的是哪一排的士兵
  volatile int nextLine = 0;
  final CyclicBarrier shiftBarrier;
  final CyclicBarrier startBarrier;

  public ShootPractice(int N, final int lineCount, int lasting) {
    this.N = N;
    this.lasting = lasting;
    this.rank = new Soldier[lineCount][N];
    for (int i = 0; i < lineCount; i++) {
      for (int j = 0; j < N; j++) {
        rank[i][j] = new Soldier(i * N + j);
      }
    }
    shiftBarrier = new CyclicBarrier(N, new Runnable() {
      @Override
      public void run() {
        // 更新下一轮打靶的排
        nextLine = (nextLine + 1) % lineCount;// 语句①
        Debug.info("Next turn is :%d", nextLine);
      }
    });
    // 语句②
    startBarrier = new CyclicBarrier(N);
  }

  public static void main(String[] args) throws InterruptedException {
    ShootPractice sp = new ShootPractice(4, 5, 24);
    sp.start();
  }

  public void start() throws InterruptedException {
    // 创建并启动工作者线程
    Thread[] threads = new Thread[N];
    for (int i = 0; i < N; ++i) {
      threads[i] = new Shooting(i);
      threads[i].start();
    }
    // 指定时间后停止打靶
    Thread.sleep(lasting * 1000);
    stop();
    for (Thread t : threads) {
      t.join();
    }
    Debug.info("Practice finished.");
  }

  public void stop() {
    done = true;
  }

  class Shooting extends Thread {
    final int index;

    public Shooting(int index) {
      this.index = index;
    }

    @Override
    public void run() {
      Soldier soldier;
      try {
        while (!done) {
          soldier = rank[nextLine][index];
          // 一排中的士兵必须同时开始射击
          startBarrier.await();// 语句③
          // 该士兵开始射击
          soldier.fire();
          // 一排中的士兵必须等待该排中的所有其他士兵射击完毕才能够离开射击点
          shiftBarrier.await();// 语句④
        }
      } catch (InterruptedException e) {
        // 什么也不做
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }

    }// run方法结束
  }// 类Shooting定义结束

  // 参与打靶训练的士兵
  static class Soldier {
    private final int seqNo;

    public Soldier(int seqNo) {
      this.seqNo = seqNo;
    }

    public void fire() {
      Debug.info(this + " start firing...");
      Tools.randomPause(5000);
      System.out.println(this + " fired.");
    }

    @Override
    public String toString() {
      return "Soldier-" + seqNo;
    }

  }// 类Soldier定义结束
}

debug和tool可以参考之前博客