ファイル処理パイプライン: Producer–Consumer

1. ファイル処理で複数スレッドをどのように調整するか

大きなファイルや複雑なデータ処理に取り組むとき、よくある課題が 作業を複数スレッドに分担する ことです。たとえば、1 つのスレッドがファイルから行を読み出し、別のスレッド群がそれらの行を処理（単語を検索、統計を集計など）します。

何が問題か？

• すべてのスレッドが同じリソース（ファイルや共有コレクションなど）を扱うと、簡単に不具合が発生します。あるスレッドが他より「先行」してしまい、レースコンディションやデータ損失、メモリ過負荷が起きる可能性があります。
• スレッドが多いのに調整がないと、あるものは待ちぼうけ、別のものは過負荷になります。

求める解決策:

• スレッド間で安全にデータを受け渡しできること。
• 処理が読み取りより遅くてもメモリあふれを起こさないこと。
• データが尽きたら、すべてのスレッドを容易に終了できること。

2. 「Producer–Consumer」パターン（プロデューサー／コンシューマー）

Producer–Consumer は、スレッド同士が互いに邪魔をせず、足並みを揃えて動けるようにする古典的なパターンです。

ここには 2 つの役割があります。Producer（プロデューサー） はデータを生成します。たとえばファイルから行を読み出したりネットワークからメッセージを受け取ったりして、それを共有キューに入れます。Consumer（コンシューマー） はそのキューからデータを取り出し、処理します（単語数を数える、DB に保存する、別のファイルへ書き込むなど）。

要点は、この 2 種類のスレッドが 独立して 動くことです。プロデューサーがコンシューマーより速く読める場合も、その逆でも、誰も無闇にブロックされません。両者の間には バッファ（キュー）があり、処理のペースを均します。

ビジュアル図

[ファイル] --(読み取り)--> [Producer] --(キューに入れる)--> [BlockingQueue] --(取り出す)--> [Consumer] --(処理する)

3. BlockingQueue による実装

Java では、この種のやり取りに BlockingQueue インターフェース（実装例: ArrayBlockingQueue）が最適です。

BlockingQueue とは？

BlockingQueue はサイズ制限付きのスレッドセーフなキューで、同期を自動的に処理してくれます。キューがいっぱいのときにプロデューサーが要素を追加しようとすると、そのスレッドは空きができるまで自動的にブロックされます。逆に、キューが空のときにコンシューマーが要素を取り出そうとすると、誰かがキューに入れるまで待機します。

この仕組みにより、レースコンディションやメモリあふれの古典的な問題が自動的に解消されます。プロデューサーは不要な要素でキューを溢れさせず、コンシューマーは空っぽを相手にせずに済みます。すべてのスレッドが落ち着いて協調して働けます。

キューの作成例

import java.util.concurrent.*;

BlockingQueue<String> queue = new ArrayBlockingQueue<>(100); // 100 要素分のバッファ

100 はキューの最大要素数です。producer がより速く動く場合、consumer が一部を処理するまで待機します。

4. スレッド協調: backpressure と終了

キューサイズの制限（backpressure）

Backpressure は、consumer が処理に追いつかないときに、producer がメモリを「食い尽くす」のを防ぐ仕組みです。

• キューが満杯なら、producer は自動的に減速します（put() がブロック）。
• キューが空なら、consumer は待機します（take() がブロック）。

これにより、スレッドの速度が異なってもシステムは安定して動作します。

終了: 「poison pill」

producer がファイルの読み取りを終えたら、consumer たちにデータの終端を知らせ、終了してよいことを伝える必要があります。

解決策:

• キューに特別なオブジェクト（「poison pill」）を入れる（例: 文字列 "__END__" など。null は不可）。
• consumer は「poison pill」を受け取ったら終了します。

consumer が複数いる場合は、その数だけ「poison pill」を入れましょう！

5. パイプライン例: ファイル読み取りと行の処理

簡単なパイプラインを実装してみましょう。

• 1 つのスレッドがファイルから行を読み取り、キューに入れる。
• 複数のスレッドがキューから行を取り出し、単語数を数えて結果を出力する。

ステップ 1. Producer — ファイル読取

import java.io.*;
import java.util.concurrent.*;

public class FileProducer implements Runnable {
    private final BlockingQueue<String> queue;
    private final File file;
    private final int consumerCount;
    private final String POISON_PILL = "__END__";

    public FileProducer(BlockingQueue<String> queue, File file, int consumerCount) {
        this.queue = queue;
        this.file = file;
        this.consumerCount = consumerCount;
    }

    @Override
    public void run() {
        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                queue.put(line); // キューが満杯なら待機
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 各 consumer 用に poison pill を投入
            try {
                for (int i = 0; i < consumerCount; i++) {
                    queue.put(POISON_PILL);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

ステップ 2. Consumer — 行処理

import java.util.concurrent.BlockingQueue;

public class LineConsumer implements Runnable {
    private final BlockingQueue<String> queue;
    private final String POISON_PILL = "__END__";

    public LineConsumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                String line = queue.take(); // キューが空なら待機
                if (POISON_PILL.equals(line)) {
                    break; // 終了
                }
                int wordCount = line.trim().isEmpty() ? 0 : line.trim().split("\\s+").length;
                System.out.println(Thread.currentThread().getName() + ": " + wordCount + " 語（行）: " + line);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

ステップ 3. Executors でパイプラインを起動

import java.io.File;
import java.util.concurrent.*;

public class PipelineDemo {
    public static void main(String[] args) throws Exception {
        int consumerCount = 3;
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);

        File file = new File("input.txt"); // あなたのファイル

        // producer を起動
        Thread producer = new Thread(new FileProducer(queue, file, consumerCount));
        producer.start();

        // consumers を ExecutorService で起動
        ExecutorService consumers = Executors.newFixedThreadPool(consumerCount);
        for (int i = 0; i < consumerCount; i++) {
            consumers.submit(new LineConsumer(queue));
        }

        // producer の終了を待つ
        producer.join();

        // consumers の終了を待つ
        consumers.shutdown();
        consumers.awaitTermination(1, TimeUnit.MINUTES);

        System.out.println("処理が完了しました！");
    }
}

6. 可視化: パイプラインの動作図

flowchart LR
    A[Producer: ファイルを読む] -- put() --> Q[BlockingQueue]
    Q -- take() --> C1[Consumer 1: 単語数を数える]
    Q -- take() --> C2[Consumer 2: 単語数を数える]
    Q -- take() --> C3[Consumer 3: 単語数を数える]
    A -.->|poison pill| Q
    Q -.->|poison pill| C1
    Q -.->|poison pill| C2
    Q -.->|poison pill| C3

7. 実用的な注意点とベストプラクティス

• キューのサイズを制限 — メモリあふれを防ぎます。
• 終了には「poison pill」を使用 — そうしないと consumers が永遠にブロックされることがあります。
• null を「poison pill」として使わない（キューに実際の null 値が入る可能性があるなら）。専用の文字列やオブジェクトを使いましょう。
• InterruptedException を処理する — スレッドを正しく終了させるために重要です。
• ExecutorService を使う — スレッドプール管理は手動でスレッドを作るより簡単で安全です。

8. パイプライン実装での典型的なミス

ミス1: 無制限のキュー → OutOfMemoryError。 LinkedBlockingQueue をサイズ無制限で使うと、consumer が追いつかない場合に producer がメモリを使い尽くしてしまいます。

ミス2: 「poison pill」を入れ忘れる → consumers がハング。 「毒薬」を入れ忘れると、コンシューマースレッドは永遠に新しいデータを待ち続けます。

ミス3: 「poison pill」が 1 つだけなのに、consumers が複数。 各コンシューマースレッドに 1 つずつ「ピル」が必要です。そうしないと全員が終了しません。

ミス4: InterruptedException を処理しない。 スレッドが割り込みされたら正しく終了処理（割り込みフラグの復元など）を行うべきで、そうしないとスレッドが「取り残される」可能性があります。

ミス5: 同期なしでスレッド間の共有変数を使う。 「車輪の再発明」はやめて、通常のリストや配列ではなく BlockingQueue を使いましょう。