【実務・中級編】Fiberを用いたバッチ処理の並列化とリソース管理：メモリとCPUの効率的な利用 – PHPコア・内部エンジンと高速化・並行処理の極意解析バイブル

Fiberを用いたバッチ処理の並列化とリソース管理：PHP内部エンジンを揺るがさないスケジューリング戦略

テックリードの私たちがコードレビューで「何千件ものレコードを処理するバッチをFiberで並列化しました」というプルリクエストを受け取った時、手放しで賞賛する前に、まず脳内でZend VMのメモリ空間とプロセスモデルの挙動をシミュレーションしなければならない。

「ただFiberを使えば高速化する」という安易な幻想は、PHPの本質を見誤っている。PHPのFiberは、Node.jsのasync/awaitやGoのgoroutineとは根本的に異なる。OSスレッドをマルチプレックスするプリエンプティブな並行処理ではなく、同一コールスタック上を協調的（コーペラティブ）にコンテキストスイッチするユーザースペースの仕組みに過ぎない。

この本質を理解せず、数千のFiberを野放図に生成すれば、Zend VMのスタックオーバーフロー、イベントループのブロッキング、そして何よりデータベースコネクションの枯渇とメモリリークという破滅的な結末が待っている。

本稿では、PHP 8.1以降で導入されたFiberを武器に、数万件の巨大バッチを安全かつ極限まで効率的に裁くための「スケジューリング戦略」と「リソース管理」の極意を、プロダクションクオリティのコードと共に伝授する。

—

1. なぜ「ナイーブなFiber実装」は本番環境で崩壊するのか

多くのエンジニアが陥る最初の罠は、すべてのデータ行に対して無思考にFiberを生成し、一斉に実行しようとすることだ。

// 【アンチパターン】絶対に真似してはいけないコード
$fibers = [];
foreach ($hugeDataSet as $data) {
$fibers[] = new Fiber(function () use ($data) {
// 重い外部API呼び出しやDBクエリ
HttpClient::post($data);
});
}
foreach ($fibers as $fiber) { $fiber->start(); }

このコードが抱える致命的な問題は以下の3点である。

1. メモリ消費量の爆発（Zend VMのスタック管理）: 各Fiberは独自のコールスタックを持つ。大量のFiberを同時生成すると、Zend VMが割り当てるヒープ・スタック領域が肥大化し、PHPの `memory_limit` を瞬時に突破する。
2. コネクションの飽和: RDB（MySQL/PostgreSQL）や外部APIの同時接続数（Max Connections）を無視してリクエストが集中するため、バックエンドのインフラストラクチャを自らDDoS攻撃する結果を招く。
3. イベントループの欠如: Fiber自体には「I/O待ちの間に他のFiberに処理を譲る」自動的なスケジューラー機能はない。ブロッキングI/O（通常の `file_get_contents` やPDOなど）をFiber内で実行した瞬間、そのスレッド全体の時間が完全に停止する。

真にスケーラブルなバッチ処理を構築するためには、「チャンク分割（バッチング）」、「イベントループによる協調的スケジューリング」、そして「同時実行数（Concurrecy）の厳格な制限」の3つを統合したアーキテクチャが不可欠となる。

—

2. 実務で耐えうる「Fiberスケジューラー」の設計

それでは、PHP単体（外部の重厚な非同期フレームワークに依存せず）で動作し、メモリとCPUを完全にコントロール下におくプロダクションレベルのスケジューラーを実装しよう。

以下のコードは、指定した最大同時実行数（Concurrency Limit）を厳守しつつ、タスクを効率的にディスパッチする非同期バッチプロセッサの完全なリファレンスである。

  • プロダクション品質のFiberベース・タスクスケジューラー

    • /
    class FiberBatchScheduler
    {
    private int $maxConcurrency;
    / @var array /
    private array $activeFibers = [];
    private array $results = [];

    public function add(callable $task): void
    {
    $this->tasks[] = $task;
    }

    public function __construct(int $maxConcurrency = 10)
    {
    $this->maxConcurrency = $maxConcurrency;
    }

    /

    • ジェネレータからタスクを受け取り、メモリ効率よくストリーミング処理する
    • @param Generator $taskGenerator

    /
    public function run(Generator $taskGenerator): array
    {
    $iterator = $taskGenerator;
    $running = true;

    while ($running || !empty($this->activeFibers)) {
    // 1. 最大同時実行数に達していない場合、新しいタスクをFiberとして投入
    while ($running && count($this->activeFibers) < $this->maxConcurrency) {
    if (!$iterator->valid()) {
    $running = false;
    break;
    }

    $task = $iterator->current();
    $iterator->next();

    $fiber = new Fiber(function () use ($task) {
    try {
    // タスクを実行し、結果を返す
    return $task();
    } catch (Throwable $e) {
    // ログ出力やエラーハンドリングをここに集約
    return [‘error’ => $e->getMessage()];
    }
    });

    // Fiberを開始
    $fiber->start();

    if (!$fiber->isTerminated()) {
    $this->activeFibers[] = $fiber;
    } else {
    // 同期的に一瞬で終わった場合の処理
    $this->results[] = $fiber->getReturn();
    }
    }

    // 2. アクティブなFiberの協調的ポーリング（簡易イベントループ）
    // 実際のI/O非同期化には Stream Select や Revolt などのイベントループを組み合わせるが、
    // ここではCPUバウンドな重い処理の並列化とメモリ枯渇を防ぐスロットリングに焦点を当てる。
    foreach ($this->activeFibers as $index => $fiber) {
    if ($fiber->isSuspended()) {
    // 非同期I/O待ちから復帰させるシグナル（今回は単純化のため再開）
    $fiber->resume();
    }

    if ($fiber->isTerminated()) {
    $this->results[] = $fiber->getReturn();
    unset($this->activeFibers[$index]);
    }
    }

    // 配列のインデックスを再採番
    $this->activeFibers = array_values($this->activeFibers);

    // CPUのスパイクを防ぐためのマイクロsleep（OSコンテキストスイッチの最適化）
    if (!empty($this->activeFibers)) {
    usleep(1000); // 1ms
    }
    }

    return $this->results;
    }
    }

    この設計における技術的ポイント

    1. ジェネレータ（`Generator`）によるメモリの定数化:
    数百万件のレコードを一度に配列としてメモリにロードするのではなく、ジェネレータを用いて「1行ずつ（またはチャンク単位で）」遅延評価してタスクを生成する。これにより、Zend VMのメモリ空間（HashTable）の消費量を数メガバイト以内に固定できる。
    2. スロットリング（`maxConcurrency`）:
    DBのコネクションプール数や外部APIのレートリミットを超えないよう、同時実行数を厳格に制限。リソースの枯渇によるコネクションエラーを物理的にシャットアウトする。
    3. GC（ガベージコレクション）への配慮:
    Fiberが終了（`isTerminated`）するたびに `$this->activeFibers` から即座にアンセットし、配列の再採番を行うことで、不要になったコールスタックのメモリ領域を速やかに解放対象にする。

    —

    3. 実践：数万件のレコード処理とリソース管理の実装例

    では、上記のスケジューラーを実際にビジネスロジックに組み込んだバッチスクリプトを見てみよう。ここでは、莫大なユーザーデータに対する外部APIへの一括同期処理を想定する。

    query(“SELECT id, email, payload FROM users WHERE status = ‘pending'”);

    // クルーソルフェッチ（PDO::FETCH_ASSOC）により、全件をメモリに乗せない
    while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
    yield function () use ($row) {
    // 各Fiber内で実行される処理
    // ※実務ではここで非同期HTTPクライアント（AmpやRevolt等）のPromiseをFiberで待ち合わせるのが理想
    $response = simulateExternalApiCall($row[‘email’], $row[‘payload’]);

    return [
    ‘id’ => $row[‘id’],
    ‘status’ => $response ? ‘synced’ : ‘failed’
    ];
    };
    }
    }

    function simulateExternalApiCall(string $email, string $payload): bool
    {
    // 外部API通信の模倣（I/O待ちのシミュレーション）
    usleep(50000); // 50ms
    return true;
    }

    // 2. 実行コンテキストの構築
    $dsn = ‘mysql:host=127.0.0.1;dbname=production_db;charset=utf8mb4’;
    $pdo = new PDO($dsn, ‘db_user’, ‘db_password’, [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_PERSISTENT => false, // Fiber環境では持続的接続の扱いに注意が必要
    ]);

    $scheduler = new FiberBatchScheduler(maxConcurrency: 15); // 同時実行数を15に制限

    $startTime = microtime(true);
    echo “=== バッチ処理開始 ===” . PHP_EOL;

    // 3. ジェネレータをスケジューラーに投入
    $taskGenerator = fetchUserDataGenerator($pdo);
    $results = $scheduler->run($taskGenerator);

    $duration = microtime(true) – $startTime;
    $successCount = count(array_filter($results, fn($r) => ($r[‘status’] ?? ”) === ‘synced’));

    echo sprintf(“=== 処理完了: 成功 %d件 / 経過時間: %.2f秒 ===”, $successCount, $duration) . PHP_EOL;

    —

    4. テックリードからの警告：運用時の落とし穴とベストプラクティス

    Fiberを用いた並行処理は強力だが、PHPという言語の特性上、以下の設計原則を破ると本番環境で深刻な障害を引き起こす。コードレビューでは必ず以下の項目をチェックしてほしい。

    ① PDOとトランザクションの共有禁止

    Fiberは同一スレッド（同一プロセス）上で実行される。もし複数のFiberが同一のPDOインスタンスを共有し、個別にトランザクション（`beginTransaction` / `commit`）を操作した場合、トランザクションの状態が混ざり合い、データ破壊や予期せぬロールバックを引き起こす。

    • 対策: データベース操作を伴うFiber並行処理を行う場合は、Fiberごとに専用のPDOコネクションを確立するか、コネクションプール層を適切に設計すること。

    ② 外部ライブラリの「ブロッキングI/O」に騙されるな

    前述の通り、Fiber内で通常の同期的なブロッキング関数（例: `file_get_contents`, 传统的cURL, `sleep()`）を呼び出すと、その瞬間に関数から制御が戻らなくなり、他のFiberの実行も完全にブロックされる。

    • 対策: 真の非同期I/Oメリットを享受したい場合は、`Revolt` イベントループや `Amp (amphp)` などのエコシステムをベースにした非同期ドライバとFiberを統合する必要がある。CPUバウンドな処理や、一定の並列度によるスループット向上（スロットリング目的）であれば、今回紹介したスケジューラーでも十分な効果を発揮する。

    ③ メモリリークの監視

    長期稼働するデーモン型バッチプロセスにおいて、Fiberのクロージャー内で外部変数を意図せずキャプチャし続ける（いわゆるクロージャーのスコープ肥大化）と、Zend VMのメモリ空間が肥大化し、OOM Killerの餌食になる。

    • 対策: バッチ処理のイテレーションごとに適切なスコープ管理を行い、不要になった変数は明示的に解放する。また、定期的にプロセスを再起動する（Workerのライフサイクル管理）設計をインフラストラクチャレベル（Supervisor等）で担保すること。

    —

    総括

    PHPにおけるFiberは、魔法の弾丸ではない。しかし、メモリの消費特性とZend VMの挙動を熟知した上で正しく設計されたスケジューラーと組み合わせれば、これまで「PHPでは重すぎて扱えなかった」大規模なバッチ処理やAPI連携を、安全かつ極めて高効率にコントロールするための強力な武器となる。

    コードの美しさは、リソースの限界を正確に把握し、それを制御する厳格な制約（ガードレール）の設計宿っている。あなたの次のバッチリファクタリングが、堅牢で美しいものになることを期待している。

    スポンサーリンク
    " "
    タイトルとURLをコピーしました