# バッチ設計書 35-messenger:consume

## 概要

本ドキュメントは、Symfony Messengerコンポーネントが提供する `messenger:consume` コマンドのバッチ設計書である。メッセージキューからメッセージを取得し、メッセージバスを通じてハンドラにディスパッチする長時間実行型ワーカープロセスの仕様を定義する。

### 本バッチの処理概要

`messenger:consume` コマンドは、指定されたトランスポート（メッセージキュー）からメッセージを消費し、メッセージバスを通じて対応するハンドラに処理をディスパッチするワーカープロセスである。

**業務上の目的・背景**：非同期メッセージング処理はモダンなWebアプリケーションにおいて重要なアーキテクチャパターンである。HTTPリクエスト中に実行すると時間がかかる処理（メール送信、画像処理、外部API呼び出し等）をメッセージキューに委任し、バックグラウンドのワーカープロセスで処理することで、レスポンスタイムの改善とシステムの信頼性向上を実現する。`messenger:consume` はこのワーカープロセスの中核を担うコマンドである。

**バッチの実行タイミング**：常時実行。Supervisorやsystemdなどのプロセスマネージャーにより管理され、アプリケーション稼働中は常にバックグラウンドで動作する。

**主要な処理内容**：
1. レシーバー（トランスポート）の解決と初期化
2. メッセージ数制限・メモリ制限・時間制限等のイベントリスナー登録
3. Workerインスタンスの生成とメッセージ消費ループの開始
4. メッセージ受信・メッセージバスへのディスパッチ
5. サービスリセット（オプション）
6. シグナルハンドリング（SIGTERM/SIGINT/SIGQUIT/SIGALRM）

**前後の処理との関連**：`messenger:setup-transports` でトランスポートのインフラストラクチャが準備されている必要がある。`messenger:stop-workers` で外部からの停止シグナルを受信できる。失敗したメッセージは失敗トランスポートに移動され、`messenger:failed:show` / `messenger:failed:retry` で管理される。

**影響範囲**：メッセージキューからのメッセージ消費、メッセージハンドラの実行、およびハンドラが行うデータベース操作やAPI呼び出し等に影響する。メッセージのAck/Reject処理によりキューの状態が変更される。

## バッチ種別

メッセージ消費（非同期メッセージ処理ワーカー）

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 常時（Supervisorなどで管理） |
| 実行時刻 | 常駐 |
| 実行曜日 | 毎日 |
| 実行日 | 毎日 |
| トリガー | プロセスマネージャー / 手動 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| トランスポート設定 | 少なくとも1つのトランスポートが設定されていること |
| トランスポートインフラ | `messenger:setup-transports` でインフラが準備済みであること |
| メッセージバス | RoutableMessageBusが設定されていること |
| イベントディスパッチャ | EventDispatcherInterfaceが利用可能であること |

### 実行可否判定

レシーバー引数または`--all`オプションにより対象トランスポートを決定する。指定されたトランスポートが存在しない場合はRuntimeExceptionが発生する。SyncTransportは自動的に除外される。`--exclude-receivers`は`--all`と併用時のみ有効。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| receivers | string[] | No | トランスポートが1つの場合はそのトランスポート | 消費するレシーバー/トランスポート名（正規表現パターン対応） |
| --limit / -l | int | No | なし | 受信メッセージ数の上限 |
| --failure-limit / -f | int | No | なし | 失敗メッセージ数の上限 |
| --memory-limit / -m | string | No | なし | メモリ使用量上限（K/M/G単位対応） |
| --time-limit / -t | int | No | なし | 時間制限（秒） |
| --sleep | int | No | 1 | メッセージがない場合のスリープ秒数 |
| --bus / -b | string | No | なし | ディスパッチ先のメッセージバス名 |
| --queues | string[] | No | なし | 消費するキュー名の制限 |
| --no-reset | bool | No | false | メッセージ処理後のサービスリセットを無効化 |
| --all | bool | No | false | すべてのレシーバーから消費する |
| --exclude-receivers | string[] | No | なし | 除外するレシーバー名（--allと併用時のみ） |
| --keepalive | int | No | 5 | キープアライブ間隔（秒） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| メッセージキュー | トランスポート依存 | 各トランスポート（AMQP、Doctrine、Redis等）からのメッセージ |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| STDOUT | テキスト | 起動メッセージ・ステータス情報 |
| STDERR | テキスト | 停止条件情報・ログ情報 |
| ログ | PSR-3 | メッセージ処理ログ（-vvオプションで詳細表示） |

### 出力ファイル仕様

ファイル出力は行わない。

| 項目 | 内容 |
|-----|------|
| ファイル名 | なし |
| 出力先 | STDOUT / STDERR / ログシステム |
| 文字コード | UTF-8 |
| 区切り文字 | なし |

## 処理フロー

### 処理シーケンス

```
1. コマンド初期化
   ├─ --keepaliveオプション処理（アラーム間隔設定）
   └─ --exclude-receiversと--allの組み合わせバリデーション
2. レシーバー解決
   ├─ --all指定時: 全レシーバーを対象
   ├─ receivers引数: 正規表現マッチングで対象レシーバー決定
   ├─ SyncTransportを除外
   └─ --exclude-receivers適用
3. イベントリスナー登録
   ├─ サービスリセットリスナー（--no-reset未指定時）
   ├─ メッセージ数制限リスナー（--limit指定時）
   ├─ 失敗数制限リスナー（--failure-limit指定時）
   ├─ メモリ制限リスナー（--memory-limit指定時）
   └─ 時間制限リスナー（--time-limit指定時）
4. Worker生成・実行
   ├─ Worker(receivers, bus, eventDispatcher, logger, rateLimiters) 生成
   ├─ sleep・queuesオプション設定
   └─ worker->run() でメッセージ消費ループ開始
5. メッセージ処理ループ
   ├─ トランスポートからメッセージ取得
   ├─ メッセージバスへディスパッチ
   ├─ ハンドラ実行
   ├─ Ack / Reject
   └─ 停止条件チェック
6. シグナルハンドリング
   ├─ SIGALRM: キープアライブ処理
   └─ SIGTERM/SIGINT/SIGQUIT: Worker停止
7. 終了
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[レシーバー解決]
    B --> C[イベントリスナー登録]
    C --> D[Worker生成]
    D --> E[メッセージ消費ループ開始]
    E --> F{メッセージあり?}
    F -->|あり| G[メッセージバスへディスパッチ]
    F -->|なし| H[sleep秒待機]
    G --> I[ハンドラ実行]
    I --> J[Ack/Reject]
    J --> K{停止条件?}
    H --> K
    K -->|未達| E
    K -->|達成| L[Worker停止]
    L --> M[バッチ終了]
```

## データベース操作仕様

### 操作別データベース影響一覧

本コマンド自体はデータベース操作を行わないが、ディスパッチされるメッセージハンドラがデータベース操作を行う場合がある。また、Doctrineトランスポート使用時はメッセージの取得・Ack・Rejectでデータベースが操作される。

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| メッセージ取得 | messenger_messages（Doctrine使用時） | SELECT | キューからメッセージを取得 |
| メッセージAck | messenger_messages（Doctrine使用時） | DELETE | 処理完了メッセージを削除 |
| メッセージReject | messenger_messages（Doctrine使用時） | DELETE | 拒否メッセージを削除 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | RuntimeException | 指定レシーバーが存在しない | 正しいレシーバー名を指定する |
| - | RuntimeException | --exclude-receiversで全レシーバーが除外された | 少なくとも1つのレシーバーを残す |
| - | InvalidOptionException | --limit が正の整数でない | 正の整数を指定する |
| - | InvalidOptionException | --time-limit が正の整数でない | 正の整数を指定する |
| - | InvalidOptionException | --exclude-receivers を --all なしで使用 | --all オプションと併用する |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | メッセージ個別のリトライ設定に依存（Messenger Retryコンポーネント） |
| リトライ間隔 | リトライ設定に依存 |
| リトライ対象エラー | ハンドラで発生した例外 |

### 障害時対応

ワーカープロセスが予期せず終了した場合は、Supervisorなどのプロセスマネージャーにより自動的に再起動される。処理中のメッセージはトランスポートの仕様に従ってリキューまたは失敗トランスポートに移動される。

## トランザクション仕様

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | メッセージ単位（ハンドラ実行とAck/Rejectが1トランザクション） |
| コミットタイミング | メッセージ処理完了時 |
| ロールバック条件 | ハンドラで例外発生時 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | メッセージ投入量に依存（--limitで上限設定可能） |
| 目標処理時間 | メッセージハンドラの処理時間に依存（--time-limitで上限設定可能） |
| メモリ使用量上限 | --memory-limitで設定可能（K/M/G単位） |

## 排他制御

同一トランスポートに対して複数のワーカープロセスを同時実行可能（トランスポートレベルのメッセージロックにより排他制御される）。`--queues` オプションで消費するキューを分離可能。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | ワーカー起動時 | 消費トランスポート名・停止条件 |
| 進捗ログ | メッセージ処理時（-vv指定） | メッセージ処理詳細 |
| シグナルログ | シグナル受信時 | 受信シグナル名・トランスポート名 |
| キープアライブログ | SIGALRM受信時（debug） | キープアライブリクエスト送信 |
| 終了ログ | ワーカー停止時 | 停止理由 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| プロセス生存 | プロセスダウン | Supervisor / プロセスマネージャー |
| メッセージ滞留数 | 環境依存 | 監視システム |
| メモリ使用量 | --memory-limit設定値 | ログ |
| 処理時間 | --time-limit設定値 | ログ |

## 備考

- 基盤実装は `src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php` に存在する
- SignalableCommandInterfaceを実装し、SIGTERM/SIGINT/SIGQUIT/SIGALRMシグナルをハンドリングする
- SIGALRMシグナルはキープアライブ処理に使用される
- レシーバー引数は正規表現パターンに対応しており、複数トランスポートのマッチングが可能
- SyncTransportは自動的に消費対象から除外される
- --no-resetオプションでサービスリセットを無効化できるが、メッセージ間でサービスの状態がリークする可能性がある
- RateLimiterが設定されている場合、レシーバーごとのレートリミットが適用される
- メモリ制限の単位変換はK（キロ）、M（メガ）、G（ギガ）、T（テラ）に対応
