【Java】CyclicBarrierによる処理同期化

Javaではマルチスレッドによる並列処理に対して同期化処理をすることができます。
同期化処理とは複数のスレッドが並行して実行されているときに処理の順番を制御することです。

同期化支援するクラスとしてCyclicBarrierというクラスが存在しています。
バリアーと呼ばれる複数スレッドが待機するポイントを処理の中で実装し、全てのスレッドがバリアーに到達すると実行されるバリアーアクションと呼ばれる処理を実行するのがCyclicBarrierの機能です。

CyclicBarrierのインスタンスを生成する際には第1引数に同期をとるスレッド数、第2引数にバリアーアクションの処理を受け取ります。

Sample.java

  public static void main(String[] arg) throws Exception {
    ExecutorService exe = Executors.newFixedThreadPool(5);

    CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {
      System.out.println("全てのスレッドの処理が完了しました。");
    });

    for(int i = 0; i < 5; i++) {
      exe.submit(new Task(cyclicBarrier));
    }
  }

上記例では5つのスレッドが全てバリアに到達したら「全てのスレッドの処理が完了しました。」と出力されるバリアーアクションを引数に渡しています。

CyclicBarrierをフィールドに持ち、並列処理が実装されているTaskクラス。

Task.java

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

public class Task implements Runnable {

  CyclicBarrier cyclicBarrier;

  public Task(CyclicBarrier cyclicBarrier) {
    this.cyclicBarrier = cyclicBarrier;
  }
  public void run() {
    System.out.println("実行されたスレッドのId = " + Thread.currentThread().getId());
    try {
      this.cyclicBarrier.await();
    } catch (InterruptedException | BrokenBarrierException e) {
      e.printStackTrace();
    }
  }
}

並列で実行される処理としては実行されているスレッドのIDを出力するだけです。

this.cyclicBarrier.await()と記載されている箇所がバリアーに当たる処理でawaitメソッドが呼ばれたスレッドは処理を中断します。全てのスレッドでawaitメソッドが呼ばれるとバリアーアクションが実行されます。

上記コードを実行すると出力される結果はこちら。

実行されたスレッドのId = 19
実行されたスレッドのId = 18
実行されたスレッドのId = 15
実行されたスレッドのId = 17
実行されたスレッドのId = 16
全てのスレッドの処理が完了しました。

5つのスレッドでIDを出力した後バリアーアクションで出力する文字列が出力されていることがわかります。