はじめに
注文サービスは、今ジョブを受け付け、後で処理する必要があります。キューを作成し、ジョブを受信して、提供されたワーカーを実行し、保存された注文を確認してからメッセージを処理済みとして確認します。
先に Lambda から DynamoDB を読み書きする と、その前提となるガイド付きラボを完了してください。この独立した VM は、ワーカー、その範囲を限定したロール、空の注文テーブル、参照データを提供します。キューやジョブは準備されていません。
認定試験との関連
このラボでは、次の試験トピックに関連する実践的な演習を行います。
- Cloud Practitioner (CLF-C02) · タスク 3.8: SQS メッセージの配信、処理、確認。
- Solutions Architect – Associate (SAA-C03) · タスク 2.1: SQS メッセージの配信、処理、確認。
- Developer – Associate (DVA-C02) · タスク 1.1: SQS メッセージの配信、処理、確認。
- DevOps Engineer – Professional (DOP-C02) · タスク 5.1: 基礎演習：SQS メッセージの配信、処理、確認。
- Solutions Architect – Professional (SAP-C02) · タスク 2.4: 基礎演習：SQS メッセージの配信、処理、確認。
独立したジョブキューを作成する
このステップでは、ジョブを送る前に空のキューを作成し、提供されたワーカーを確認します。
Terminal の隣で AWS View を使い、現在のキュー、ワーカーの結果、保存された注文を比較します。無関係な参照データを保持してください。
Amazon Simple Queue Service（SQS） は、コンシューマー向けのジョブをメッセージとして保存します。プロデューサーはジョブを送り、コンシューマーはジョブを処理します。キューは両者のタイミングを分離し、プロデューサーがコンシューマーの処理完了を待つ必要をなくします。Standard キューは、メッセージを複数回配信する場合があります。受信と処理済みとしての確認は、別の操作です。
準備された作業ディレクトリから始めます。
cd /home/labex/project
提供されたワーカーと空の注文テーブルを確認します。ワーカーは、商品 1 個あたり 250 セントに 100 セントの手数料を加えて計算します。その実行ロールは注文テーブルだけに書き込めます。参照テーブルは、保持すべき無関係なデータです。
aws lambda get-function-configuration --function-name labex-q01-worker --query '{Name:FunctionName,Role:Role,Runtime:Runtime}'
aws dynamodb scan --table-name labex-q01-orders --query Items
空のアイテム一覧が返るはずです。自分のキューを作成します。--query QueueUrl --output text は、そのアドレスをプレーンテキストとして選択します。$(...) は、後のコマンドで使うために、その出力をシェル変数 QUEUE_URL に保存します。
QUEUE_URL=$(aws sqs create-queue --queue-name labex-q01-jobs --attributes VisibilityTimeout=300 --query QueueUrl --output text)
可視性タイムアウトは、受信したメッセージが再び利用可能になるまで、コンシューマーに 300 秒を与えます。これは一時的に見えなくするもので、削除ではありません。次のラボで、期限切れと再配信を学びます。
キューの識別情報と現在の件数を確認します。
aws sqs get-queue-attributes --queue-url "$QUEUE_URL" --attribute-names QueueArn VisibilityTimeout ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible
VisibilityTimeout は 300、両方のメッセージ件数は 0 となるはずです。件数は概算の運用上の指標であり、業務処理完了の保証ではありません。AWS View では、キューに利用可能なジョブも処理中のジョブも 0 件と表示されます。提供された Lambda には実行がなく、注文テーブルも空のままです。
公式コンソールは、CLI で確認するものと同じキュー名、型、URL、ARN を表示します。これらは例の値です。引き続き自分の QUEUE_URL を使ってください。このラボのリソースは AWS View で観察します。
出典：AWS SQS。
注文ジョブを送信して受信する
このステップでは、ジョブを送信・受信して、利用可能なメッセージと処理中のメッセージの違いを観察します。
受信、処理、処理済みとしての確認の順で進めます。保存された注文を確認してから、現在のレシートハンドルを使ってキューメッセージを削除してください。
メッセージの本文はアプリケーションデータです。SQS は JSON をテキストとして保持し、コンシューマーがそれを解釈する必要があります。小さな合成注文ジョブを 1 件送ります。
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"id":"queue-order","quantity":2}'
レスポンスには MessageId と MD5OfMessageBody が含まれます。メッセージ ID はメッセージを識別し、MD5 は本文のバイト列を要約します。これはキューによる受け付けを確認するものであり、注文の完了ではありません。AWS View には利用可能なジョブが 1 件表示されますが、注文テーブルは空のままです。
メッセージを 1 件受信し、レスポンスを保存します。> は、出力を表示する代わりに received.json にリダイレクトします。--wait-time-seconds 5 は、すぐに利用できるジョブがない場合に、短いロングポーリングの待機を行えます。
aws sqs receive-message --queue-url "$QUEUE_URL" --max-number-of-messages 1 --wait-time-seconds 5 --message-system-attribute-names ApproximateReceiveCount --output json > received.json
JSON からフィールドを選択する jq で、安全なアプリケーション本文と受信回数を読み取ります。
jq '.Messages[0] | {MessageId,Body,Attributes}' received.json
本文には queue-order と数量 2 が含まれ、最初の受信では受信回数が 1 となるはずです。完全なレスポンスには、レシートハンドルも含まれます。これは、この特定の配信試行に対応する値です。このメッセージを処理済みとして確認するときには、最新のハンドルを使います。
aws sqs get-queue-attributes --queue-url "$QUEUE_URL" --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible
利用可能な件数は 0、非表示の件数は 1 となるはずです。AWS View ではジョブは処理中（in flight）ですが、注文はまだありません。受信だけでは処理も削除も行われません。300 秒以内に次のステップへ進んでください。読むのにそれ以上かかった場合は、処理と削除の前に同じファイルへもう一度受信し、最新のレシートハンドルを取得してください。
この実際の例は、処理前の配信を示します。自分の作業ディレクトリではメッセージ ID が異なります。
処理済みとして確認する前にジョブを処理する
このステップでは、受信した本文を処理し、永続化された注文を確認してから、メッセージを処理済みとして確認します。
実際に受信した本文をワーカーの入力に使います。fromjson は、SQS レスポンス内の JSON 文字列を JSON オブジェクトに変換し、リダイレクトがそのオブジェクトを job.json に書き込みます。
jq '.Messages[0].Body | fromjson' received.json > job.json
提供された Lambda は、この注文オブジェクトを受け付けます。Lambda コースと同様に、fileb://job.json はファイルのバイト列を送り、末尾に指定したファイルが関数のレスポンスを受け取ります。ワーカーを呼び出します。
aws lambda invoke --function-name labex-q01-worker --payload fileb://job.json worker-response.json
Invoke API のステータス成功だけでは、処理成功を証明できません。レスポンス本文を確認します。
cat worker-response.json
processed: true、数量 2、total_cents: 600 が返るはずです。その後、関数が返したメッセージだけに頼らず、保存されたアイテムを読み取ります。
aws dynamodb get-item --table-name labex-q01-orders --key '{"id":{"S":"queue-order"}}' --consistent-read --query Item
queue-order、数量 2、合計 600 が返るはずです。AWS View には、実際のワーカーの入力と結果、および同じ永続化された注文が表示されます。処理済みとして確認するまで、ジョブは処理中のままです。ワーカーのレスポンスか保存アイテムのどちらかが誤っている場合は、削除せずに診断のためにメッセージを保持してください。
注文を確認した後、現在のレシートハンドルをプレーンテキストとして選択し、その配信をキューから削除します。
RECEIPT_HANDLE=$(jq -r '.Messages[0].ReceiptHandle' received.json)
aws sqs delete-message --queue-url "$QUEUE_URL" --receipt-handle "$RECEIPT_HANDLE"
削除が成功すると、通常は何も表示されません。件数をもう一度読み取ります。
aws sqs get-queue-attributes --queue-url "$QUEUE_URL" --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible
両方の件数が 0 となるはずです。AWS View には空のキューと 1 件の保存された注文が表示されます。キューによる受け付け、一時的な配信、業務処理、処理済みとしての確認を分けられました。実際のアプリケーションでは Standard キューがメッセージを再配信する可能性があり、この手順はエンドツーエンドで厳密に 1 回の処理を保証するものではありません。後のユニットで、永続的な重複保護を学びます。
自分のキューと結果だけを後片付けする
このステップでは、提供されたリソースを保持しながら、自分のキュー、結果、ログを削除します。
提供されたワーカー、テーブル構造、参照データを保持しながら、キューと作成した注文を削除します。キューの削除は、残るすべてのジョブを破棄するため、先に前のステップが成功したことを確認してください。
aws sqs delete-queue --queue-url "$QUEUE_URL"
aws dynamodb delete-item --table-name labex-q01-orders --key '{"id":{"S":"queue-order"}}'
ワーカーの実行によって、CloudWatch Logs グループが作成されました。後片付けの一部として、このラボの実行ログを削除します。
aws logs delete-log-group --log-group-name /aws/lambda/labex-q01-worker
成功するサービスクエリでリソース状態を確認します。
aws sqs list-queues
aws dynamodb scan --table-name labex-q01-orders --query Items
aws dynamodb scan --table-name labex-q01-reference --query Items
キュー URL は残らず、注文アイテム一覧は空になり、参照アイテムには引き続き keep unchanged がある必要があります。AWS View にはキュー、注文、実行ログがなく、提供されたワーカーと参照リソースは残ります。認証エラーやネットワークエラーは、削除成功の証拠ではありません。
リソース確認後、通常のレスポンスファイルを削除します。
rm -f received.json job.json worker-response.json
環境を終了する前に、このステップの検証を使ってください。
まとめ
SQS Standard キューを作成し、JSON 注文ジョブを送信・受信して、実際の Lambda 処理と永続化された DynamoDB アイテムを確認し、レシートハンドルでメッセージを処理済みとして確認しました。受け付け、処理中の配信、業務処理の完了を区別し、その後、提供されたリソースを保持しながら、自分のキュー、注文、実行ログを削除しました。
