Amazon SQSワーカーのアーキテクチャーをLambdaイベントソース/EventBridge Pipes/独自の3パターンで比較してみた

quiver

2024.04.18

メッセージキューはプロデューサー・コンシューマー方式で非同期に通信するメッセージング･パターンです。

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等)をプラガブルに指定します。

自由度高くコンシューマーを定義できるため、イベントソースやターゲットを変更する余地を残すなど、柔軟性を担保したい場合にもおすすめです。

ターゲットはバッチでわたってくるメッセージの処理ロジックだけを記述します。

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数に応じてスケーリングする処理を実装します。