【Java】CyclicBarrierでスレッド同期を行う
今回はスレッド同期の際に使えるCyclicBarrierクラスについてみていきます。
CyclicBarrierは指定したスレッド数分が待機状態に入るまで、スレッドの停止ができます。
宣言の例
CyclicBarrier barrier = new CyclicBarrier(待機スレッド数);
awaitメソッドでスレッドを待機させます。
barrier.await();
試しにマルチスレッドのプログラムを実装し動作を確認してみましょう。
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class Test {
//3スレッドのCyclicBarrier
static CyclicBarrier barrier = new CyclicBarrier(3);
//テスト用数値
static int num = 0;
public static void main(String[] args) throws InterruptedException {
//結果表示スレッド
new Thread(Test::runResult).start();
//数値加算スレッド
new Thread(Test::runIncrement).start();
//スレッド一時停止
Thread.sleep(1000);
//数値加算スレッド
new Thread(Test::runIncrement).start();
}
//加算処理
static void runIncrement() {
try {
num++;
//スレッド一時停止
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}
//結果表示処理
static void runResult() {
try {
//スレッド一時停止
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(num);
}
}
実行結果
2
上記プログラムでは待機数3のCyclicBarrierを構築し、それぞれのスレッドで加算処理、結果表示処理を行います。
先に結果表示を行うスレッドを起動し、スレッド一時停止も行った上で加算スレッドを実行しましたが、加算処理が終了した結果が表示されています。
awaitによって各種スレッドが停止できていることがわかりますね。
CyclicBarrierはすべての指定スレッド数が待機状態になった時に特定処理を実行できる仕組みがあります。
CyclicBarrier barrier = new CyclicBarrier(待機スレッド数, Runnableオブジェクト);
で使用できます。
Runnableオブジェクトは最後のスレッドによって処理が実行されます。
次のプログラムで内容を確認してみましょう。
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class Test {
//3スレッドのCyclicBarrier（指定数待機完了時runResultを呼び出す指定）
static CyclicBarrier barrier = new CyclicBarrier(3, Test::runResult);
//テスト用数値
static int num = 0;
public static void main(String[] args) throws InterruptedException {
//数値加算スレッド
new Thread(Test::runIncrement, "スレッド1").start();
//一時停止
Thread.sleep(500);
//数値加算スレッド
new Thread(Test::runIncrement, "スレッド2").start();
//一時停止
Thread.sleep(500);
//数値加算スレッド
new Thread(Test::runIncrement, "スレッド3").start();
}
//加算処理
static void runIncrement() {
try {
num++;
//スレッド一時停止
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}
//結果表示処理
static void runResult() {
System.out.println(Thread.currentThread().getName());
System.out.println(num);
}
}
実行結果
スレッド3 3
最後に待機されるであろう、スレッド3で待機完了時の結果が表示されたのがわかりますね。
CyclicBarrierはresetメソッドによって待機中のスレッドの実行を再開できます。
resetメソッドを呼び出す際、BrokenBarrierExceptionがスローされます。
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class Test {
//3スレッドのCyclicBarrier
static CyclicBarrier barrier = new CyclicBarrier(3);
public static void main(String[] args) throws InterruptedException {
new Thread(Test::runIncrement, "スレッド1").start();
new Thread(Test::runIncrement, "スレッド2").start();
Thread.sleep(500);
System.out.println("待機スレッド数="+barrier.getNumberWaiting());
barrier.reset();
}
//加算処理
static void runIncrement() {
try {
//スレッド一時停止
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+":終了");
}
}
実行結果
待機スレッド数=2 java.util.concurrent.BrokenBarrierException at java.base/java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:252) at java.base/java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:364) at Test.runIncrement(Test.java:21) at java.base/java.lang.Thread.run(Thread.java:833) スレッド1:終了 java.util.concurrent.BrokenBarrierException at java.base/java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:252) at java.base/java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:364) at Test.runIncrement(Test.java:21) at java.base/java.lang.Thread.run(Thread.java:833) スレッド2:終了
resetメソッドが呼び出された後は再利用が可能になります。
スレッド同期の方法はsynchronized、Object.wait等その他にもいくつか方法はありますが、こんなクラスがあったとは驚きました。
もしかしたら特定のアプリケーションで役に立つかもしれませんね。
多分使わないだろうなぁ。