【Java学習｜豆知識】Phaserで始める、柔軟で強力なJava並行処理！

皆さん、こんにちは！Javaエンジニアの〇〇です。

今回は、Javaの並行処理ライブラリの中でも、特に柔軟性と拡張性に優れた`java.util.concurrent.Phaser`について、その魅力を余すところなくお伝えしたいと思います。

1. Phaserはなぜ重要か？どんな課題を解決するか？

並行処理の世界では、複数のタスクが協調して動作する必要があります。従来の`CountDownLatch`や`CyclicBarrier`は、固定された数のスレッドを待機させるのには便利ですが、実行中のスレッド数が変動する場合や、より複雑な同期が必要な場合には対応が難しいという課題がありました。

`Phaser`は、この課題を解決するために登場しました。`Phaser`は、「フェーズ」という概念を持ち、各フェーズで参加者の数を動的に変更したり、完了条件を柔軟に設定したりすることができます。これにより、より複雑な並行処理シナリオでも、シンプルかつ効率的に同期を取ることが可能になります。

例えば、以下のようなシナリオで`Phaser`が役立ちます。

段階的な処理: 複数のワーカーが連携して、複数の段階（フェーズ）を経て処理を進める場合。各フェーズの完了後に、次のフェーズに進む前に同期を取りたい。
動的な参加者: 処理の途中で参加するスレッドが増減する場合。
複雑な依存関係: あるタスクの完了が、複数の他のタスクの開始条件となっている場合。

2. Phaserの基礎知識

`Phaser`は、Javaの`java.util.concurrent`パッケージに含まれる同期ユーティリティです。その核となる概念は「フェーズ」と「パーティ」です。

フェーズ (Phase): `Phaser`は、番号付けされたフェーズ（0から始まります）で進行します。各フェーズは、一連の同期ポイントを表します。
パーティ (Party): `Phaser`に参加するエンティティ（通常はスレッド）を「パーティ」と呼びます。各フェーズの完了は、そのフェーズに参加しているすべてのパーティが「登録解除（deregister）」または「待機（awaitAdvance）」するまで待機されます。

`Phaser`の主な操作は以下の通りです。

`register()`: `Phaser`に新しいパーティを登録します。
`arriveAndAwaitAdvance()`: 現在のパーティが現在のフェーズの完了を通知し、次のフェーズに進むまで待機します。
`arriveAndDeregister()`: 現在のパーティが現在のフェーズの完了を通知し、以降のフェーズから登録を解除します。
`arrive()`: 現在のパーティが現在のフェーズの完了を通知しますが、次のフェーズに進むまで待機しません。
`awaitAdvance(int phase)`: 指定されたフェーズが完了するまで待機します。

`Phaser`は、`onAdvance(int phase, int registeredParties)`というメソッドをオーバーライドすることで、各フェーズの完了時にカスタムロジックを実行することも可能です。このメソッドは、次のフェーズに進む前に呼び出され、フェーズの進行を制御したり、追加の処理を行ったりするのに役立ちます。

3. Phaserの実装/解決策

ここでは、`Phaser`を使って、複数のワーカーが段階的に処理を進めるシナリオを実装してみましょう。

まず、`Phaser`インスタンスを作成し、初期のパーティ数を指定します。次に、各ワーカー（スレッド）は、それぞれのタスクを実行した後、`arriveAndAwaitAdvance()`を呼び出して次のフェーズに進みます。

もし、あるワーカーがそのフェーズの処理を完了したら、それ以降のフェーズには参加しない、という場合は`arriveAndDeregister()`を使用します。

さらに、各フェーズの完了時に特定の処理を行いたい場合は、`Phaser`を継承して`onAdvance()`メソッドをオーバーライドします。

4. サンプルプログラム

以下のサンプルプログラムは、3つのワーカーが3つのフェーズで処理を行い、各フェーズの完了時に`Phaser`が同期を取る様子を示しています。

import java.util.concurrent.Phaser;

public class PhaserExample {

public static void main(String[] args) {
// 3つのパーティ（ワーカー）でPhaserを初期化
Phaser phaser = new Phaser(3);

// ワーカー1
new Thread(new Worker(“Worker 1”, phaser), “Worker 1 Thread”).start();
// ワーカー2
new Thread(new Worker(“Worker 2”, phaser), “Worker 2 Thread”).start();
// ワーカー3
new Thread(new Worker(“Worker 3”, phaser), “Worker 3 Thread”).start();

// メインスレッドもPhaserに参加（必要に応じて）
// phaser.register(); // メインスレッドも参加させる場合
}

static class Worker implements Runnable {
private final String name;
private final Phaser phaser;

public Worker(String name, Phaser phaser) {
this.name = name;
this.phaser = phaser;
}

@Override
public void run() {
System.out.println(name + ” が処理を開始しました。”);

// フェーズ0: 初期処理
doPhase(0);
// フェーズ0完了を通知し、フェーズ1に進むまで待機
int currentPhase = phaser.arriveAndAwaitAdvance();
System.out.println(name + ” がフェーズ ” + (currentPhase – 1) + ” を完了しました。”);

// フェーズ1: 中間処理
doPhase(1);
// フェーズ1完了を通知し、フェーズ2に進むまで待機
currentPhase = phaser.arriveAndAwaitAdvance();
System.out.println(name + ” がフェーズ ” + (currentPhase – 1) + ” を完了しました。”);

// フェーズ2: 最終処理
doPhase(2);
// フェーズ2完了を通知し、このワーカーは登録解除
phaser.arriveAndDeregister();
System.out.println(name + ” がフェーズ ” + (currentPhase – 1) + ” を完了し、終了しました。”);
}

private void doPhase(int phase) {
System.out.println(name + ” がフェーズ ” + phase + ” の処理を実行中です…”);
try {
// 処理のシミュレーション
Thread.sleep((long) (Math.random() 1000));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}

このプログラムを実行すると、各ワーカーがそれぞれのフェーズの処理を行い、`Phaser`によって同期されながら次のフェーズに進む様子が確認できます。

5. 応用・注意点

`onAdvance()`の活用: `Phaser`を継承して`onAdvance()`メソッドをオーバーライドすることで、各フェーズの完了時に全体の状況をチェックしたり、次のフェーズの開始条件を動的に設定したりできます。例えば、`onAdvance()`で`return true`を返すと、`Phaser`はterminated状態になり、それ以降のフェーズ進行は停止します。
登録解除のタイミング: `arriveAndDeregister()`は、そのワーカーが以降のフェーズに参加しない場合にのみ使用します。もし、そのワーカーが後続のフェーズにも参加する必要があるのに登録解除してしまうと、同期が取れなくなり、デッドロックの原因となる可能性があります。
`Phaser`の再利用: `Phaser`は、一度完了したフェーズをリセットして再利用することも可能です。これは、バッチ処理などで同じ処理フローを複数回繰り返す場合に便利です。
`Phaser`とExecutorService/Virtual Threads: `Phaser`は、`ExecutorService`やVirtual Threadsと組み合わせることで、さらに強力な並行処理基盤を構築できます。タスクの実行管理と同期を効果的に分離することが可能です。
デバッグの難しさ: 並行処理は、その性質上、デバッグが難しい場合があります。`Phaser`を使用する際は、各スレッドの実行順序や同期の状態を注意深く確認することが重要です。ログ出力やデバッガを活用して、意図した通りに動作しているかを確認しましょう。

`Phaser`は、その柔軟性から、様々な並行処理の課題に対応できる強力なツールです。ぜひ、皆さんのプロジェクトでも活用してみてください！