非同期キューは、高コストなタスクの完了をWebリクエストが待たないようにしますが、ジョブ間の競合を単独で解決するものではありません。インポート、再処理、キャンペーンによって数千件のメッセージが作成され、すべてのコンシューマーを占有したときに問題が生じます。注文確認、在庫引当、アカウントのロック、トランザクション通知の送信といった即時影響のあるアクションが、待機可能な作業の後ろに回されます。
PHPキューの優先順位を管理することは、メッセージに数値フィールドを追加するだけではありません。これは、ビジネスフローを反映し、制約のある依存先を保護し、負荷増加時にも予測可能な動作を維持すべきアーキテクチャ上の判断です。
影響、期限、コストで作業を分類する
キューを作成する前に、非同期ジョブの一覧を作成してください。各ジョブについて、開始者、使用する依存先、通常の所要時間、ビジネス上の期限、遅延時の影響を特定します。緊急性は必ずしも重要性と同義ではありません。たとえば、財務照合は非常に重要でも数時間の待機を許容できる場合があり、決済検証は実行時間が短くても迅速な応答を必要とする場合があります。
有用な分類には通常、4つのサービスクラスがあります。
- クリティカル: 金銭、セキュリティ、一貫性、または即時のコミットメントを保護するアクション。非常に短い目標待機時間と予約済みの処理能力が必要です。
- インタラクティブ: ユーザーが開始する作業、またはアプリケーションから要求されたドキュメント生成など、ほぼリアルタイムの体験を完了するために必要な作業。
- 遅延: 定期同期、集計、インデックス更新など、必要ではあるものの即時の期限がないタスク。
- バルク: インポート、移行、再インデックス、キャンペーン、再処理。ほかに負荷がない場合でも、その量またはコストゆえに処理速度の制限が必要です。
ジョブごとのコストも記録してください。レート制限のあるAPIを呼び出す、負荷の高いクエリを実行する、または大きなファイルを処理するメッセージは、短いローカル更新と同じように競合させるべきではありません。サービスクラスは、期限と、その作業がシステムに与える負荷の種類を表す必要があります。
真の分離が必要ならキューを分ける
ジョブの実行特性が均一で、同じ依存先を使用し、トランスポートが信頼できる優先順位を提供する場合は、優先順位付きの単一キューで足りることがあります。しかし、取得順序だけでは処理能力を保証しません。すでに実行中のバルクジョブは、worker、接続、または外部クォータを引き続き占有します。
次のいずれかの制約がある場合はキューを分けてください。
- クリティカルジョブとバルクジョブで、待機時間の目標が明確に異なる。
- あるジョブ種別が、決済API、メール、ERPなど、脆弱またはレート制限のある依存先にアクセスする。
- 実行時間に大きな差があり、長時間ジョブがプロセスを長く保持する。
- デプロイ、一時停止、再試行、スケーリングを個別に制御する必要がある。
- あるフローのエラーまたは異常な入力が、別のフローを劣化させてはならない。
PHPアプリケーションでは、たとえばcritical、interactive、deferred、bulkのような明示的なキューへメッセージをルーティングするパターンが最も読みやすいものです。メッセージングコンポーネントにはSymfony Messenger、Laravel Queues、または選択したブローカーとの独自統合を使用できます。この原則はフレームワークに依存しません。キュー内の優先順位は、類似ジョブを並べ替えるためにこの分離を補完できますが、互換性のないクラス間の分離を置き換えるものではありません。
予約済み処理能力と最大同時実行数を定義する
クラスごとにコンシューマーを割り当て、運用上の最小値と最大値の両方を設定してください。クリティカルキューには、インポートに吸収されない処理能力が必要です。一方、バルクキューには、データベース、CPU、ストレージ、外部プロバイダーを飽和させないための最大同時実行数が必要です。
すべてのworkerを、クリティカルに絶対優先で全キューから読み取るように設定することは避けてください。この方法では、予約済みコンシューマーが他のジョブを取得できなければ処理能力が遊休になる可能性があり、ルールなしに取得できれば飢餓を引き起こす可能性があります。実用的な代替案は、次を組み合わせることです。
- クリティカルとインタラクティブ専用のworker。
- クォータに従って遅延とバルクを処理する共有worker。
- 総プロセス数だけでなく、依存先種別ごとの制限。
- CPU使用率だけでなく、キュー深度とメッセージ経過時間に基づくスケーリング。
正しい数値は一律ではありません。データベースとAPIが許容する同時実行数、観測された所要時間、各クラスの目標待機時間を基に決める必要があります。
飢餓を防ぎ、バックプレッシャーを適用する
緊急なものを優先しても、遅延作業が完了しなくてよいことにはなりません。クリティカルメッセージが常にある場合、厳格な優先順位ポリシーは飢餓を生む可能性があります。つまり、低優先度ジョブが無期限に滞留します。限られた数のクリティカルメッセージの後に一定量の遅延メッセージを処理する、または緊急でない作業のために処理能力の小さな割合を予約するなど、測定可能なfairnessルールを定めてください。
このルールは依存先の制約を尊重しなければなりません。クリティカルとバルクが高コストなロックを伴って同じテーブルに書き込む場合、両方を並行実行するとレイテンシーが悪化する可能性があります。その場合、クォータは共有リソースに適用するか、ジョブをより小さいバッチに再設計することが適切です。
バックプレッシャーは、完了可能な量を超える作業が流入したときに発生します。workerを無制限に増やしても解決しません。対応方法を定義してください。
- インポートのサイズ、頻度、または同時実行数を送信元で制限する。
- バッチを再開可能な単位に分割し、同時に発行する数を制御する。
- キューまたは依存先が閾値を超えた場合、明示的なスケジューリングで遅延作業を延期する。
- 即時再試行ではなく、一時停止と遅延再試行によってレート制限応答を尊重する。
- 処理のために受け付けられた操作と、実際に完了した操作をプロダクトに通知する。
受け付けと実行を区別することが重要です。インポートを受信したと返しても、直ちに開始できることを意味しません。この透明性により、技術的な変更が即時利用可能性の約束と解釈されることを防げます。
再試行、低速処理、冪等性を制御する
再試行は処理能力を消費し、偶発的に優先度の高い負荷となる可能性があります。エラーを一時的なものと恒久的なものに分類してください。一時的なネットワーク障害は、待機時間を増やし時間的分散を加えた再試行を正当化する場合があります。一方、無効なバリデーション、存在しないリソース、失効した認証情報は、無限に繰り返すのではなく、レビューのための経路に送るべきです。
ジョブ種別ごとに最大実行時間を設定してください。低速なjobがworkerを無期限に保持してはなりません。分割できる場合は、進捗を記録する独立したメッセージでページ、ファイル、またはセグメントを処理します。分割できない場合は、厳格な制限、安全なキャンセル、上限超過ジョブをレビューする手順を使用してください。
プロデューサーがメッセージを再送したり、コンシューマーが外部APIの呼び出し後に失敗したりすると、優先順位によって影響の重複リスクが高まります。冪等なhandlerを設計してください。安定した操作キーを使用し、遷移の状態を永続化し、2回処理しても1回処理した場合と同じビジネス上の効果になるようにします。ブローカーの重複排除は重複を減らせますが、アプリケーションや外部統合における冪等性の代わりにはなりません。
キューサイズだけでなく待機を観測する
最も古いメッセージが長く待機している場合や、コンシューマーが継続的に失敗している場合、キューが短くても問題が隠れている可能性があります。サービスクラスごとに、最も古いメッセージの経過時間、発行から開始までの時間、実行時間、エラー率、再試行、レビューに送られたジョブを測定してください。
これらのメトリクスを、アクティブな同時実行数、深度、流入・流出率、接続使用量、依存先の応答時間、受信したレート制限で補完してください。最も有用なシグナルは、クリティカルの待機が目標を超える、クォータが制限されている間にバルクキューが増大する、再試行がトラフィックを支配する、または別クラスのピーク中に予約済み処理能力が遊休状態になることです。
固定のメッセージ数だけでなく、傾向とサービス目標に基づいてアラートを設定してください。インポートにおける1,000メッセージは正常な場合があります。一方、数分間待機している注文確認に属する10メッセージは重大である可能性があります。
共有フローの例と導入チェックリスト
緊急注文、通知、カタログのバルクインポートを処理するプラットフォームを想定してください。注文はcriticalへ、トランザクション通知はinteractiveへルーティングされ、インポートはbulkへ送信されるページに分割されます。注文用workerには予約済み処理能力があります。インポートは同時実行数が制限され、データベースレイテンシーが増加すると処理速度を落とします。通知は、遅延再試行によりプロバイダーのクォータを尊重します。処理が繰り返されても、操作キーにより2つの在庫引当や2回の状態変更送信を防ぎます。
このモデルを既存アプリケーションに導入するには、次の手順を実施します。
- handlerを棚卸しし、期限、影響、依存先に基づいてサービスクラスを割り当てる。
- ルーティングを変更する前に、所要時間、待機時間、エラーを測定する。
- まずクリティカルフローとバルクフローを分離し、最小処理能力を予約する。
- 依存先ごとの同時実行数制限とバックプレッシャーポリシーを定義する。
- ビジネス上の効果を冪等にし、再試行と実行時間を制限する。
- 新しい振り分けを有効化する前に、負荷急増、プロバイダー障害、大量入力をテストする。
- クォータとクラスを定期的に見直す。優先順位はプロダクトとともに変化するビジネスポリシーです。
目指す結果は、すべてを優先扱いにすることではなく、大量処理をビジネスのほかの部分のブロックに変えることなく、各ジョブに一貫した処理能力と期限を与えることです。
