SQSを利用する

AWS

Amazon SQS、通称SQSは公式で以下の表現をしています。

マイクロサービス、分散システム、およびサーバーレスアプリケーション用の完全マネージド型メッセージキュー

ref. https://aws.amazon.com/jp/sqs/

これだけみると正直ピンと来なかったのですが、時間がかかる処理を非同期で行えるようにキューとして使えるサービスです。

今回の流れ

今回作るものは以下のような流れになります。

sequenceDiagram
    autonumber
    participant C as Client
    participant A as API Gateway
    participant P as 受付Lambda
    participant Q as SQS
    participant E as Event Source Mapping
    participant W as Worker Lambda

    C->>A: POST /jobs
    A->>P: 直接呼び出し (Push)
    P->>P: job_idを生成
    P->>Q: SendMessage(job_id)
    Q-->>P: 受付成功
    P-->>A: statusCode 202 + job_id
    A-->>C: HTTP 202 Accepted + job_id

    Note over C,W: HTTPレスポンスはここで完了

    E->>Q: ポーリング (Pull)
    Q-->>E: 1件のメッセージ
    E->>W: Workerを呼び出す
    W->>W: 5秒の処理
    W-->>E: 正常終了

要点は以下です。

  • 受付Lambdaでjob_idを作り、SQSへ送る
  • Event Source MappingはSQSをポーリングし、Worker Lambdaを呼び出す
  • Worker LambdaはSQSメッセージを受け取り、本体の処理を行う

受付Lambda、Worker Lambdaは、ここで役割を区別するためにつけた名前です。

ポイントは、HTTPレスポンスを返した後にWorker Lambdaの処理が行われることです。

受付Lambda

job_idはHTTPレスポンス、SQSメッセージ、Workerログに入れて、別々の実行を紐づけるために用意しました（無くても動きます）。

コードとしては以下のようになります。

job_id = str(uuid.uuid4())

sqs_client.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps({"job_id": job_id}),
)

return {
    "statusCode": 202,
    "body": json.dumps({"job_id": job_id, "status": "accepted"}),
}

send_messageで指定したSQSにメッセージを渡し、202を返しています。

202 Acceptedは、リクエストを受け付けたものの、処理はまだ完了していないことを表すステータスコードです。

ref. https://www.rfc-editor.org/rfc/rfc9110.html#name-202-accepted

SQS

まずAWS CDKのSQSのコードです。

job_queue = sqs.Queue(
    self,
    "JobQueue",
    # Worker Lambdaのタイムアウト（10秒）の6倍にするのがAWS推奨の様子
    # ref. https://docs.aws.amazon.com/lambda/latest/dg/services-sqs-configure.html
    visibility_timeout=Duration.seconds(60),  # Worker Lambdaが処理しているメッセージを一時的に他の処理から見えなくする時間
    retention_period=Duration.days(1),
    removal_policy=RemovalPolicy.DESTROY,
)

SQSにはStandardキューとFIFOキューがありますが、今回はStandardキューを使います。デフォルトではStandardキューが作られます。

FIFOキューは順序保証が必要な場合に使うもので、fifo=Trueを指定して作ります。

受付Lambdaから送られたメッセージは、Worker Lambdaが処理できるまでSQSに保存されます。

Event Source Mapping と Worker Lambda

Event Source MappingはLambda側のAWSリソースです。SQSをポーリングするものが何かしら必要になるのですが、LambdaサービスがSQSをポーリングするため、Workerにポーリングコードは不要でした。

使用するには次のような設定を行います。Worker Lambdaと合わせて以下のようになります。

# 5秒の疑似処理が途中でタイムアウトしないよう、余裕を持たせる
worker_timeout = Duration.seconds(10)

worker_function = lambda_.Function(
    self,
    "WorkerFunction",
    runtime=lambda_.Runtime.PYTHON_3_13,
    handler="handler.lambda_handler",
    code=lambda_.Code.from_asset(str(functions_root / "worker")),
    timeout=worker_timeout,
)

worker_function.add_event_source(
    lambda_event_sources.SqsEventSource(
        job_queue,
        batch_size=1,
        enabled=True,
    )
)

batch_size=1としているため、Worker Lambdaは最大1件のメッセージを受け取ります。 enabled=TrueはEvent Source Mappingを有効にする設定です。無効にした場合、メッセージはSQSに残りますが、Worker Lambdaは呼び出されません。

このようにして、Event Source MappingがSQSをポーリングし、取得したメッセージを使ってWorker Lambdaを同期的に呼び出します。

ref. https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html

ref. https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html

Worker LambdaではSQSメッセージからjob_idを取り出します。今回は重い処理の代わりに5秒待ち、処理前後にjob_idをログへ出します。

このログはCloudWatch Logsで確認します。HTTPレスポンスとWorker Lambdaのログに同じjob_idが出ることで、同じジョブであることを確認できます。

job_id = json.loads(record["body"])["job_id"]

# Worker Lambdaの処理開始をCloudWatch Logsへ記録する
print(
    json.dumps(
        {"event": "job_started", "job_id": job_id},
        ensure_ascii=False,
    )
)

# HTTPレスポンスがWorkerの完了を待たないことを確認するため、
# 重い処理の代わりに5秒待つ
time.sleep(5)

# Worker Lambdaの処理完了をCloudWatch Logsへ記録する
print(
    json.dumps(
        {"event": "job_completed", "job_id": job_id},
        ensure_ascii=False,
    )
)

CDK

改めて全体像を描くと以下のようになります。

# 5秒の疑似処理が途中でタイムアウトしないよう、余裕を持たせる
worker_timeout = Duration.seconds(10)

job_queue = sqs.Queue(
    self,
    "JobQueue",
    # AWSの推奨に従い、Worker Lambdaのタイムアウトの6倍にする
    visibility_timeout=Duration.seconds(60),
    retention_period=Duration.days(1),
    removal_policy=RemovalPolicy.DESTROY,
)

producer_function = lambda_.Function(
    self,
    "ProducerFunction",
    runtime=lambda_.Runtime.PYTHON_3_13,
    handler="handler.lambda_handler",
    code=lambda_.Code.from_asset(str(functions_root / "producer")),
    timeout=Duration.seconds(5),
    environment={"QUEUE_URL": job_queue.queue_url},
)
job_queue.grant_send_messages(producer_function)

worker_function = lambda_.Function(
    self,
    "WorkerFunction",
    runtime=lambda_.Runtime.PYTHON_3_13,
    handler="handler.lambda_handler",
    code=lambda_.Code.from_asset(str(functions_root / "worker")),
    timeout=worker_timeout,
)
worker_function.add_event_source(
    lambda_event_sources.SqsEventSource(
        job_queue,
        batch_size=1,
        enabled=True,
    )
)

これまで触れてない点としては、

job_queue.grant_send_messages(producer_function)によって、受付LambdaにSQSへメッセージを送る権限を付与します。

動作確認

確認したいこと

HTTPレスポンスがWorker Lambdaの5秒待機より先に返ることを確認します。

実行手順

※省略していますが、実際には他にもリージョンの確認などが入ります。

uv sync --extra dev
# 対象のアカウントとリージョンで初めてCDKを使う場合だけ実行する
uv run -- npx --yes aws-cdk@2 bootstrap
uv run -- npx --yes aws-cdk@2 deploy --outputs-file cdk-outputs.json

先にWorker Lambdaのログを表示しておきます。

WORKER_LOG_GROUP=$(uv run python -c 'import json; print(json.load(open("cdk-outputs.json"))["AsyncApiWithSqsStack"]["WorkerLogGroupName"])')
aws logs tail "$WORKER_LOG_GROUP" --since 5m --format short --follow

別のターミナルからAPIへリクエストします。

API_URL=$(uv run python -c 'import json; print(json.load(open("cdk-outputs.json"))["AsyncApiWithSqsStack"]["ApiUrl"])')
curl -i -X POST -w '\ntime_total=%{time_total}\n' "$API_URL"

実測結果

ap-northeast-1で実行しました。

HTTP/2 202
{"job_id": "624e17d7-e6db-411b-8cfd-8856bb59412c", "status": "accepted"}
time_total=3.299863

Worker Lambdaのログは以下でした。

23:16:25 {"event": "job_started", "job_id": "624e17d7-e6db-411b-8cfd-8856bb59412c"}
23:16:30 {"event": "job_completed", "job_id": "624e17d7-e6db-411b-8cfd-8856bb59412c"}

HTTPレスポンスは約3.3秒で返り、Worker Lambdaの完了を待たずにレスポンスがあったことを確認できました。

Worker Lambdaはjob_startedの5秒後にjob_completedを出し、正常終了しました。レスポンスとログのjob_idも一致しています。

クリーンアップ

忘れずに確認後はリソースを削除します。

uv run -- npx --yes aws-cdk@2 destroy