【ITニュース解説】How We Made Video Processing 10x Faster with Symfony Messenger

今日のアプリケーションに求められるのは、ユーザーが快適に利用できる速さと、どのような操作にも素早く反応する応答性である。しかし、動画の処理やレポートの生成、大きなファイルのアップロードといった、時間やコンピューターの資源を大量に消費するタスクを扱う場合、これを実現するのは簡単なことではない。ユーザーを何秒も待たせることなく、これらの重いタスクを裏側で処理する仕組みが必要となる。この解決策が「非同期処理」であり、特にSymfonyフレームワークの「Messenger」コンポーネントは、この種の課題を解決するための強力なツールとして注目されている。

従来のWebアプリケーションでは、ユーザーのリクエストを受けてから、そのリクエストに関連する全てのアクションが完了するまで、Webサーバーは応答を返すのを待っていた。これを「同期処理」と呼ぶ。例えば、ユーザーが動画をアップロードし、その動画のエンコード処理がサーバー上で行われる場合、このエンコードが完了するまでユーザーのWebブラウザは「読み込み中」の状態になり、アプリケーションが停止したかのように感じてしまう。これはユーザー体験を著しく損なう。さらに、Webサーバーが一度に処理できるリクエストの数には限りがあるため、もし数件の重い処理がサーバーを占有してしまうと、他のユーザーからのリクエストを処理できなくなり、アプリケーション全体のパフォーマンスが低下し、スケーラビリティ（拡張性）も制限されてしまう。

このような同期処理の課題を解決するのが、非同期処理を可能にするSymfony Messengerである。Messengerは、アプリケーションの異なる部分を切り離し、独立して動作させるための「メッセージバス」という仕組みを提供する。これにより、Webアプリケーションは、動画処理などの重いタスク自体を実行する代わりに、そのタスクの内容を記した「メッセージ」をメッセージバスに「送信」するだけで済む。このメッセージは一時的に保存され、その後、バックグラウンドで動いている別の「ワーカー」と呼ばれるプログラムがメッセージを「受信」し、実際の処理を実行する。この一連の流れは、Webアプリケーションがタスクの指示を出す「生産者（Producer）」となり、ワーカーがその指示を受け取って処理する「消費者（Consumer）」となる「生産者・消費者パターン」に基づいている。

Symfony Messengerを理解するためのいくつかの重要な概念がある。一つは「メッセージ」であり、これはタスクを実行するために必要なデータだけを保持するシンプルな情報のかたまりである。例えば、動画ファイルのパスやユーザーIDなどがこれにあたる。次に「メッセージハンドラー」があり、これは特定のメッセージを受け取った際に実行される具体的なビジネスロジック（実際の処理内容）を記述する部分である。これらのメッセージを適切なハンドラーにルーティングするのが「メッセージバス」の役割である。また、メッセージを一時的に保存し、ワーカーに配送する役割を担うのが「トランスポート」である。記事ではRabbitMQ、Redis、SQSなどが例として挙げられているが、これらはメッセージキューと呼ばれる種類の技術で、メッセージを確実に届け、処理の負荷を分散させるために使われる。そして、これらのトランスポートからメッセージを継続的に読み取り、対応するハンドラーを呼び出してタスクを実行するのが「コンシューマー」または「ワーカー」と呼ばれるバックグラウンドプロセスである。さらに、同じメッセージを複数の独立したグループのワーカーに同時に処理させる「コンシューマーグループ」という機能もあり、動画の圧縮と音声の文字起こしといった複数の処理を並行して行うような、より複雑なワークフローにも対応できる。

実際に、4K動画のような非常に大きなファイルをアップロードするシナリオを考えてみよう。もしこの動画のエンコード処理を、アップロードと同時に一つの同期リクエスト内で完結させようとすると、ユーザーは処理が完了するまで長時間待たされることになる。これは明らかに劣悪なユーザー体験であり、Webサーバーの資源も独占されてしまう。そこで、Symfony Messengerを用いた非同期かつ並列な動画処理が有効となる。このアプローチでは、まずアップロードされた動画ファイルを小さな「チャンク」（断片）に分割する。そして、各チャンクを処理するための個別のメッセージを作成し、それらをメッセージキューに送信する。

具体的な実装では、まずSymfony Messengerコンポーネントと、大量のメッセージを扱うのに適したメッセージブローカー（メッセージキューのサーバー）であるRabbitMQをアプリケーションに導入する。次に、設定ファイルでRabbitMQをトランスポートとして定義し、どのメッセージがどのトランスポートに送られるかをルーティングルールとして設定する。動画処理の例では、初期の動画分割をトリガーするProcessVideoMessageと、分割された各動画チャンクを処理するProcessVideoChunkMessageという二種類のメッセージを作成する。それぞれのメッセージには、対応する「メッセージハンドラー」が用意される。ProcessVideoMessageHandlerは、受け取った動画ファイルを複数のチャンクに分割し、それぞれのチャンクについてProcessVideoChunkMessageをメッセージバスにディスパッチする。そしてProcessVideoChunkMessageHandlerは、個々のチャンクに対してエンコードや透かしの挿入といった実際の動画処理を実行する。

ユーザーが動画をアップロードした直後、Webコントローラーは非常に高速な処理として、最初のProcessVideoMessageをメッセージバスにディスパッチする。これにより、Webアプリケーションはすぐにユーザーに応答を返し、「動画処理がバックグラウンドで開始されました」といったメッセージを表示できる。ここからが非同期処理の真髄である。事前に起動しておいた複数のワーカーが、RabbitMQのキューからProcessVideoChunkMessageをそれぞれ独立して取得し、割り当てられたチャンクの処理を開始する。例えば、10個のワーカーを起動すれば、10個のチャンクが同時に処理される。一つのワーカーがチャンクの処理を終えると、すぐに次のチャンクをキューから引き出し、処理を続けるため、すべてのワーカーが常に忙しく動作し、動画全体の処理が劇的に高速化される。本番環境では、Supervisordのようなプロセス管理ツールを使って、これらのワーカープロセスを自動的に起動・管理するのが一般的である。

すべての動画チャンクの処理が完了したら、それらを元の1つの動画ファイルに結合する必要がある。この最終的な結合処理もまた、時間のかかるタスクであるため、非同期で実行することが望ましい。この段階では、「イベント」という仕組みを活用する。各ワーカーが個々の動画チャンクの処理を成功裏に終えた際、直接結合処理を指示するのではなく、VideoChunkProcessedEventというイベントメッセージをディスパッチする。このイベントは「動画チャンクの処理が一つ終わった」という出来事を知らせるものである。

アプリケーション内には、このイベントを「監視」し、動画全体の進捗を追跡する専用のサービス（例えばChunkTrackerService）を用意する。このサービスは、データベースなどを利用して、どの動画の何番目のチャンクが処理済みであるかを記録していく。そして、処理済みのチャンク数が、元々あったチャンクの総数と一致した時点で、「全てのチャンクの処理が完了した」と判断し、最終的な結合処理を指示するCombineVideoChunksMessageをメッセージバスにディスパッチする。このCombineVideoChunksMessageに対応するハンドラーは、VideoCombinerサービスを利用して、処理済みのすべてのチャンクファイルを一つの動画ファイルに結合する。この結合処理にはFFmpegという強力なツールが使われ、チャンクを再エンコードすることなく結合できるため、非常に高速に完了する。これにより、ユーザーは動画のアップロードから最終的な結合まで、一連の重い処理が全てバックグラウンドで行われ、アプリケーションは常に高速で応答性が高い状態を保つことができる。

この非同期プログラミングパターンをSymfony Messengerと共に採用することで、二つの大きなメリットが得られる。一つは、アプリケーションの各コンポーネントが互いに強く依存することなく、独立して機能するようになることである。これにより、コードのモジュール化が進み、アプリケーションの設計がより分かりやすくなり、将来の変更やメンテナンスが容易になる。もう一つは、巨大なデータセットを小さな部分に分割し、複数のワーカーで並行して処理することで、大量のデータを非常に高いパフォーマンスで処理できる点である。このアプローチによるスケーラビリティは、サーバーの物理的な資源や運用戦略によってのみ制限され、理論上はほぼ無限に拡張できる。