高水準シンクロナイザ

1. CountDownLatch: 合図でスタート

マルチスレッドの世界では、スレッドのグループが足並みをそろえて動くことがよく必要になります。全員が一緒に開始・終了したり、次の段階に進んだりするためです。例えば:

レースを想像してください。車はスタートラインで待機中 — すでにエンジンを温め終えた車もあれば、まだタイヤを確認している車もあります。けれど審判が旗を振るまでは誰も動きません。これがまさに「調整」の仕事です。

別の例：友達と夕食の準備をしているとします。誰かは野菜を刻み、誰かはお湯を沸かし、誰かは塩がどこへ行ったか探しています。大事なのは、調理を始める前に全員が下準備を終えていることです。

このような場合に、Java は安全で分かりやすく、wait() と notify() の苦痛とは無縁の同期ツールを提供します。その中でも便利なのが CountDownLatch。カウンタ式の「錠」のように動き、ゼロになるまでは「扉」は閉じたまま誰も先へ進めません。全員がチェックインするとラッチが開き、スレッドはそろって次へ進みます。

CountDownLatch

CountDownLatch は「一度きりのバルブ」で、1つ以上のスレッドが、他のスレッドが所定回数の処理を終えるまで待機できるようにします。

マラソンのスタートのようなものです。全ランナーはスタートラインでピストルの合図を待ちます。審判が発砲した瞬間（カウントダウンがゼロになった）に全員が走り出します。

どう動くのか

CountDownLatch はスレッドのためのスターターホイッスルのようなものです。生成時に数（例：3）を指定します。これはレースが始まる前に必要な「3つの合図」に相当します。

開始を待つスレッドは await() を呼びます。スタートラインでいつでも飛び出せる準備はできていますが、まだブレーキを踏んで待機です。準備をする他のスレッドは、準備できるたびに countDown() を呼びます。まるで「準備完了！」という合図を出すかのように。

カウンタがゼロに達したら——バン！——待っていたスレッドが同時にスタートします。

ただし注意: CountDownLatch は一度きりです。カウンタがゼロになったあとで元に戻すことはできません。リボルバーではなく、打ち上げ花火のようなものです。一度鳴ったらおしまい。

例: N 個のタスクの完了を待つ

import java.util.concurrent.CountDownLatch;

public class LatchDemo {
    public static void main(String[] args) throws InterruptedException {
        int workers = 3;
        CountDownLatch latch = new CountDownLatch(workers);

        for (int i = 1; i <= workers; i++) {
            int id = i;
            new Thread(() -> {
                System.out.println("ワーカー " + id + " の作業を開始");
                try { Thread.sleep(500 + id * 200); } catch (InterruptedException ignored) {}
                System.out.println("ワーカー " + id + " の作業が終了");
                latch.countDown(); // カウンタを減らす
            }).start();
        }

        System.out.println("メインスレッドはすべてのワーカーの終了を待っています...");
        latch.await(); // 全ワーカーが終わるのを待つ
        System.out.println("全ワーカーが終了！ メインの処理を続行します。");
    }
}

出力:

メインスレッドはすべてのワーカーの終了を待っています...
ワーカー 1 の作業を開始
ワーカー 2 の作業を開始
ワーカー 3 の作業を開始
ワーカー 1 の作業が終了
ワーカー 2 の作業が終了
ワーカー 3 の作業が終了
全ワーカーが終了！ メインの処理を続行します。

例: 合図による同時スタート

CountDownLatch startSignal = new CountDownLatch(1);

for (int i = 0; i < 5; i++) {
    new Thread(() -> {
        try {
            System.out.println(Thread.currentThread().getName() + " は開始を待っている");
            startSignal.await(); // 合図を待つ
            System.out.println(Thread.currentThread().getName() + " がスタート！");
        } catch (InterruptedException ignored) {}
    }).start();
}

Thread.sleep(1000);
System.out.println("スタートの合図！");
startSignal.countDown(); // 全スレッドが同時にスタート

2. CyclicBarrier: 複数回のフェーズとバリアアクション

CyclicBarrier: 集合場所で合流

CyclicBarrier はスレッドの集合場所です。各スレッドはそれぞれのルートを進み、作業を行い、その後「バリア」で合流します。山で焚き火の場所に集まるようなイメージです。全員が集まったらバリアが開き、グループはそろって先へ進みます。

CountDownLatch との主な違いは、何度でも使えることです。各集合のあとに「再装填」され、チームは次のステージへ進めます。

想像してみてください: ハイカーのグループが長いルートを進んでいます。各自のペースで歩き、誰かは蝶を撮影し、誰かは Wi‑Fi を探しています。しかし、各峠では焚き火の場所で合流し、お互いを待ってから次の行き先を決めます。これが CyclicBarrier の動作です。

どう動くのか

バリアを作成し、何人の参加者が集まるべきか（例：4）を指定します。各スレッドはチェックポイントに到達したら await() を呼び、他のスレッドを待ちます。4人全員が集まると、バリアが「カチッ」と開いて全員が解放されます。

「バリアアクション」も指定できます。グループが集まったときにちょうど1回だけ実行される小さなコードです。たとえば焚き火を起こす、あるいは「ステージ完了、次へ進む」とログに記録するなど。これにはコンストラクタに Runnable を渡します。

重要: 一度きりの CountDownLatch と違い、CyclicBarrier は何度でも使えます。各「集合」のあと、次のステージに備えて再び使える状態になります。

例: フェーズの同期

import java.util.concurrent.CyclicBarrier;

public class BarrierDemo {
    public static void main(String[] args) {
        int parties = 3;
        CyclicBarrier barrier = new CyclicBarrier(parties, () -> {
            System.out.println("全員がバリアに到達！ 新しいフェーズを開始します。");
        });

        for (int i = 1; i <= parties; i++) {
            int id = i;
            new Thread(() -> {
                try {
                    System.out.println("スレッド " + id + " はフェーズ1で作業中");
                    Thread.sleep(300 + id * 200);
                    System.out.println("スレッド " + id + " はバリアを待機");
                    barrier.await(); // 残りを待つ

                    System.out.println("スレッド " + id + " はフェーズ2で作業中");
                    Thread.sleep(200 + id * 100);
                    System.out.println("スレッド " + id + " はバリアを待機 (2)");
                    barrier.await(); // 再び待つ

                    System.out.println("スレッド " + id + " は作業を完了");
                } catch (Exception e) {
                    System.out.println("エラー: " + e);
                }
            }).start();
        }
    }
}

出力:

スレッド 1 はフェーズ1で作業中
スレッド 2 はフェーズ1で作業中
スレッド 3 はフェーズ1で作業中
スレッド 1 はバリアを待機
スレッド 2 はバリアを待機
スレッド 3 はバリアを待機
全員がバリアに到達！ 新しいフェーズを開始します。
スレッド 1 はフェーズ2で作業中
...

バリアアクション

コンストラクタに CyclicBarrier のアクション（Runnable）を渡すと、全スレッドがバリアに到達したときに1回だけ実行されます（状態更新、ログ出力など）。

落とし穴: 1つのスレッドが落ちたら？

1つのスレッドが例外を投げたりバリアに到達しなかった場合、他のスレッドは永遠に待つか、BrokenBarrierException を受け取ります。バリアは「壊れ」るため、再作成が必要です。

この章は、より生き生きとした比喩や会話調に書き換えることもできます。たとえば前章の「オーケストラ」の流れに自然につながるようにするとよいでしょう。

3. Phaser: 大規模コンサートの名指揮者

Phaser はいわば「スーパーバリア」です。CountDownLatch と CyclicBarrier の長所を兼ね備えつつ、さらに柔軟です。まるでオーケストラのように、楽章の合間に奏者が出入りしても、指揮者は各楽章が全員の準備完了で始まるようにコントロールします。

通常のバリアと異なり、Phaser は段階的（フェーズ）に動作します。フェーズが次々に切り替わり、最初の楽章だけ演奏する人、後から参加する人、途中で退場する人がいても、Phaser は平然と扱えます。

どう動くのか

まず Phaser を作成し、通常は参加者数（parties）を指定します。各スレッドは登録（register()）し、自分のパートを実行し、フェーズの終わりに arriveAndAwaitAdvance() を呼んで完了を知らせ、他の参加者を待ちます。全員が到達すると Phaser は次のフェーズへ切り替わり、これを繰り返します。

もう不要になった参加者は、arriveAndDeregister() で「一礼して」退場できます。逆に、新しい参加者は演奏の途中でも register() で参加できます。

いつ Barrier より Phaser が良いか

Phaser を選ぶべきケース:

• スレッド数が動的に変わる。
• 複数のフェーズがあり、すべての参加者がすべてに参加する必要はない。
• 手作業の同期に煩わされず、最大限の柔軟性が欲しい。

要するに Phaser は指揮者です。バトンを振るだけでなく、オーケストラの編成、楽章数、遅刻や早退にまで適応します。

例: 動的なスレッド数での段階処理

import java.util.concurrent.Phaser;

public class PhaserDemo {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(1); // メインスレッド

        for (int i = 1; i <= 3; i++) {
            phaser.register(); // 参加者を登録
            int id = i;
            new Thread(() -> {
                for (int phase = 1; phase <= 2; phase++) {
                    System.out.println("スレッド " + id + " はフェーズ " + phase + " で作業中");
                    try { Thread.sleep(200 + id * 100); } catch (InterruptedException ignored) {}
                    phaser.arriveAndAwaitAdvance(); // 残りを待つ
                }
                System.out.println("スレッド " + id + " は作業を完了");
                phaser.arriveAndDeregister(); // phaser から離脱する
            }).start();
        }

        // メインスレッドもフェーズに参加
        for (int phase = 1; phase <= 2; phase++) {
            phaser.arriveAndAwaitAdvance();
            System.out.println("メインスレッド: フェーズ " + phase + " が完了");
        }
        phaser.arriveAndDeregister();
        System.out.println("すべてのフェーズが完了！");
    }
}

特徴:

• 参加者をその場で追加・削除できる。
• 現在のフェーズ番号を取得可能: phaser.getPhase()。
• phaser を終了できる: phaser.forceTermination()。

4. Exchanger: スレッド間でデータを交換する

Exchanger<T> は2つのスレッド間でデータを交換するためのシンクロナイザです。各スレッドが exchange(data) を呼び、2つが出会ったときに互いのデータを交換します。

たとえ: 2人の配達員が交差点で出会い、荷物を交換するイメージです。

どう動く?

• 一方のスレッドが exchange(data1) を呼ぶ — もう一方を待つ。
• もう一方が exchange(data2) を呼ぶ — 双方が相手のデータを受け取る。
• 相手が来なければ待機（タイムアウト指定も可）。

例: producer と consumer のバッファ交換

import java.util.concurrent.Exchanger;

public class ExchangerDemo {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();

        // Producer
        new Thread(() -> {
            String data = "producer からのデータ";
            try {
                System.out.println("Producer: データを送信中");
                String response = exchanger.exchange(data);
                System.out.println("Producer: 応答を受信: " + response);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // Consumer
        new Thread(() -> {
            try {
                String received = exchanger.exchange("consumer からの応答");
                System.out.println("Consumer: データを受信: " + received);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

出力:

Producer: データを送信中
Consumer: データを受信: producer からのデータ
Producer: 応答を受信: consumer からの応答

用途:

• スレッド間のバッファ交換（例：一方がファイルから読み、もう一方がネットワークへ書く）。
• 2つのスレッド間でのフェーズ同期。

5. 実践: 並列パイプライン処理

課題: ゲームの「ティック」（フェーズ）

複数のスレッドがあり、それぞれがゲーム世界の一部（例：物理、AI、レンダリング）を担当しているとします。各「ティック」（フェーズ）で同期し、ずれが起きないようにしたい。

解決: CyclicBarrier または Phaser を使います。

import java.util.concurrent.CyclicBarrier;

public class GameTickDemo {
    public static void main(String[] args) {
        int subsystems = 3;
        CyclicBarrier barrier = new CyclicBarrier(subsystems, () -> {
            System.out.println("すべてのサブシステムがティックを終了。次を開始します。");
        });

        for (int i = 1; i <= subsystems; i++) {
            int id = i;
            new Thread(() -> {
                for (int tick = 1; tick <= 5; tick++) {
                    System.out.println("サブシステム " + id + " はティック " + tick + " で作業中");
                    try { Thread.sleep(100 + id * 50); } catch (InterruptedException ignored) {}
                    try {
                        barrier.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

課題: 大量ワーカーのための「バルブ」

準備のあとに同時スタートすべき 100 本のワーカースレッドがあるとします（例：負荷テスト）。

解決: CountDownLatch を使います。

import java.util.concurrent.CountDownLatch;

public class MassStartDemo {
    public static void main(String[] args) throws InterruptedException {
        int workers = 100;
        CountDownLatch ready = new CountDownLatch(workers);
        CountDownLatch start = new CountDownLatch(1);

        for (int i = 0; i < workers; i++) {
            new Thread(() -> {
                System.out.println("スレッドはスタート準備完了");
                ready.countDown(); // 準備完了を通知
                try {
                    start.await(); // 共通の合図を待つ
                    System.out.println("スレッドがスタート！");
                } catch (InterruptedException ignored) {}
            }).start();
        }

        ready.await(); // すべてのスレッドの準備完了を待つ
        System.out.println("全員準備完了！ スタート！");
        start.countDown(); // スタートの合図を出す
    }
}

6. シンクロナイザ利用時の典型的なミス

エラー №1: CountDownLatch を再利用可能なバリアとして使ってしまう。
CountDownLatch は一度きりです！ ゼロに達したあと「再装填」できません。再利用可能なフェーズには CyclicBarrier または Phaser を使いましょう。

エラー №2: 例外（InterruptedException, BrokenBarrierException）を処理していない。
await() は例外を投げることがあります。必ず処理しないと、スレッドが「固まる」か、エラーで終了するおそれがあります。InterruptedException と BrokenBarrierException に注意してください。

エラー №3: 1つのスレッドがバリアに到達しない。
1本でもスレッドが落ちたり await() を呼ばなかったりすると、他は永遠に待機します（または BrokenBarrierException を受け取ります）。全参加者が確実にバリアへ到達するようにしましょう。

エラー №4: Phaser で deregister() を忘れる。
スレッドが作業を終えたのに arriveAndDeregister() を呼ばないと、Phaser は「死んだ」参加者を待ち続けます。必ず正しく Phaser から外してください。

エラー №5: Exchanger を2つを超えるスレッドで使う。
Exchanger は2スレッド間の交換専用です。スレッドが2つより多いとデッドロックになります。

エラー №6: 仕組みを理解せずに異なるシンクロナイザを混在させる。
同じグループのスレッドに対して、複数種類のバリアやラッチを同時に使うのは避けましょう。混乱やハングアップの原因になります。