メッセージキューはプロデューサー・コンシューマー方式で非同期に通信するメッセージング･パターンです。
AWSが提供するメッセージキュー Amazon SQS が大昔の2006年から本稼働していることからわかるように、decoupleの代名詞とも言えるSQSはクラウドネイティブなアーキテクチャーにおいて最も重要なコンポーネントの一つです *1。
※ https://aws.amazon.com/message-queue/ から引用
最近のAWS事情を反映して、Amazon SQSのコンシューマーのアーキテクチャを以下の3パターンで検討する機会がありましたので紹介します。
- Lambdaイベントソース
- EventBridge Pipes
- 独自実装
コンシューマーの選び方
システムがシンプルで見通しが良いほど、運用は楽です。
サーバーレスでマージドなサービスにポーリングやスケーリングをオフロードできると、作り込む余地が少なくなり、運用費も安くなって一石二鳥です。
SQSと相性のよいLambdaでコンシューマーを実装できるかが大きな分かれ目です。
Lambda単体でメッセージを処理することもあれば、システム連携のためにLambdaを呼び出すこともあるでしょう。
いずれにせよ、メッセージ処理が15分を超えることがあったり、10GBを超える潤沢なメモリが必要だったり、GPUが必要だったりすると、実行環境にLambdaを採用できません。
Lambdaを採用できる場合、シンプルさを追求するならLambdaイベントソース、ソース･ターゲットの選択肢に含みを残したり、メッセージの前処理など柔軟に設定したいならPipesがオススメです。あとから切り替えるのも、比較的軽微な設定変更とコード変更ですみます。
ワーカーにLambdaよりも強力なコンピュートサービス(Amazon ECS等)が必要だったり、Step Functionsのような他のAWSサービスでメッセージを処理したい場合は、Pipesを検討しましょう。
システム･時代ごとに散らかっているコンシューマー実装方法を集約させたい場合にも、助けになるでしょう。
Pipesでも対応が難しい場合は、サービスを組み合わせて独自実装しましょう。
| 方式 | 運用の手間 | 運用費 | プラガブル | 実行基盤の制約 | ポーリング | スケーリング |
|---|---|---|---|---|---|---|
| イベントソース | 低 | 低 | 中 | 高 | イベントソース | イベントソース |
| Pipes | 低 | 低 | 高 | 中 | Pipes | Pipes |
| 独自実装 | 高 | 中 | 低 | 低 | アプリケーション | AutoScaling等 |
各実装方式の詳細は以下のとおりです
1. Lambdaイベントソース型
Lambdaのイベントソースマッピングを利用すると、SQS・Kinesis Data Streamなど様々なソースにポーリングを行い、Lambdaを呼び出せます。
イベントソースが暗黙にポーリングやSQSメッセージの受信･削除をするため、Lambdaアプリケーションは受け取ったメッセージの処理ロジックだけを記述します。
def lambda_handler(event, context):
for message in event['Records']:
print(message['body'])覚えることが少なくて済み、管理も楽なため、SQSコンシューマーの初手としておすすめできる一方で、最大の問題点はLambdaの制約に縛れることです。
メッセージ処理が15分を超えたり、10GBを超える潤沢なメモリが必要だったり、GPUが必要だったりすると、Lambda(とLambdaに依存するイベントソース)を採用できません。
ポーリングに伴うAPI呼び出しの利用費も発生しますが、コンシューマー処理の利用費のほとんどはLambdaでしょう。
2. EventBridge Pipes型
Lambdaではコンピューティングリソースが足りず、Amazon ECSやAWS Batchを使いたい、あるいは、Step Functionsなど他のAWSサービスと連携したい場合におすすめなのが、2022年末に登場したEventBridge Pipesです。
Pipesはコンシューマーを
- ソース
- ポーリング
- ターゲット
の3つに明示的に分割し(美しいですね!) *2、ソース(SQS等)とターゲット(Lambda等)をプラガブルに指定します。
自由度高くコンシューマーを定義できるため、イベントソースやターゲットを変更する余地を残すなど、柔軟性を担保したい場合にもおすすめです。
ターゲットはバッチでわたってくるメッセージの処理ロジックだけを記述します。
SQS-Pipes-Lambdaのターゲット実装
def lambda_handler(event, context):
for message in event:
print(message['body'])Lambdaイベントソースと同じく、ポーリングに伴うAPI呼び出しやPipes起動の利用費も発生しますが、コンシューマー処理の利用費のほとんどはターゲットのコンピューティングリソースが占めることになるでしょう。
次のドキュメントにあるように、ターゲットによってバッチサイズが異なります。
Amazon EventBridge Pipes のバッチ処理と同時実行 - Amazon EventBridge
例として、SQSのコンシューマーをLambdaとECS Fargateでざっくりと比較します。 ECS Fargateをターゲットに指定すると、バッチサイズは1に固定されてメッセージ単位でタスクが起動します。FargateはイメージのPULLも含めた処理時間が分単位に切り上げて課金されます。 大量の軽量なメッセージを処理するようなユースケースでは、ECS FargateよりもLambdaに分があるでしょう。
実際に構築して負荷をかけて評価すると、机上検討では見えていなかったいろいろな制約が見えてきます。
3. 独自実装型
EC2やECSでコンシューマーを常駐させ、アプリケーションでポーリング･メッセージ処理などを実装します。
import boto3
sqs = boto3.client('sqs')
SQS_QUEUE_URL = 'https://sqs.eu-central-1.amazonaws.com/123456789012/bar'
response = sqs.receive_message(
QueueUrl = SQS_QUEUE_URL,
)
for message in response.get("Messages", []): # メッセージの受信
print(message["Body"]) # 処理
sqs.delete_message(
QueueUrl = SQS_QUEUE_URL,
ReceiptHandle = message['ReceiptHandle']
) # メッセージの削除同時に、Auto ScalingやCloudWatch AlarmといったAWSサービスを組み合わせ、visible message数に応じてスケーリングする処理を実装します。
※ 図はAWSの参照実装"Scaling based on Amazon SQS"から
要件に合わせて自由にカスタマイズできる一方で、自分たちで作り込んだサーバーフルなAWSリソース群の面倒を見ないといけないため、運用負荷が重くのしかかります。スケーラビリティや可用性を確保しつつ、コストも意識した絶妙な舵取りも求められます。
エンジニアのチカラの見せ所でもありますが、個人的には、マネージドサービスにオフロードして、意識したくない領域です。
メッセージキューのコンシューマー処理で実装をサボってはいけないこと
SQSに限らず、メッセージキューのコンシューマー処理でサボりたくなるけれども、サボると後で痛い目に合うのが、冪等性とデッドレターキュー(DLQ)です。
忘れずに実装しましょう。
冪等性
SQSに限らず、デフォルトでは同じメッセージが複数回配信される可能性(at least once)のあるメッセージキューは少なくありません。
同じメッセージを複数回処理しても副作用が生じないように、冪等(idempotent)に実装しましょう。
SQS FIFOのように、一度しか配信されないこと(exactly-once)を保証するキューもありますが、スループットなど、様々なトレードオフを伴います。
デッドレターキュー
キューイングされたにも関わらず、コンシューマーの不具合などにより処理されなかった(できなかった)メッセージが生まれることがあります。
そのようなメッセージを特別なキュー(デッドレターキュー/DLQ)に送信することで、正常系メッセージを滞りなく処理し続け、異常系メッセージを個別に対応できるようになります。
責任分界点から SQS のコンシューマー実装を考える
過去には、AWSの進化に伴い、コンシューマーの処理がEC2→Lambdaイベントソース→EventBridge Pipesと疎結合になっていくことを、同様の3分類で行いました。
参考まで。
4
