弁当の注文が表に載るまで
お客さんが、注文のファイルを S3 バケットにアップロードします。しばらくすると、その注文が DynamoDB の表の1行になっています。間で、誰もボタンを押していません。
小さな3つの段階が働きました。イベント、キュー、関数です。このレッスンでは、1つのアップロードを追い、段階が失敗したり、2回動いたりしたときの様子を見ます。
このレッスンのゴール
メモ
1つのアップロードが、キューと関数を通って表に届くまでを、たどることができる。
キーワード
| ことば | いみ | れい |
|---|---|---|
| パイプライン | 仕事を、次へ次へと渡していく段階の列 | |
| イベント | 何かが起きた、という知らせ | |
| メッセージ | キューで順番を待つ、1つの文字列 | |
| 項目 | DynamoDB の表に入る、1行 | |
| 冪等 | 2回実行しても、同じ結果になること | 新しいときだけ項目を置く |
学生が入力するセルは、このプレビューでは空白です。
各段階が、文字列を次へ渡す
アップロードがあると、S3 がイベントを送ります。1つの規則が、そのイベントを SQS のキューにメッセージとして入れます。関数がメッセージを読み、表に項目を書きます。
キューが運ぶのは、辞書ではなく文字列です。だから、イベントは JSON の文字列にされ、関数が元の辞書に戻します。ここのイベントは、文書にある形で、スタックが作ったものではありません。
実行: イベントが body になる
import json
# S3 のイベントの、文書にある形。読む部分だけに削ってあります。
event = {"Records": [{
"eventName": "ObjectCreated:Put",
"s3": {"bucket": {"name": "bento-uploads"},
"object": {"key": "orders/sato/3.json"}}}]}
# キューが運ぶのは文字列なので、イベントは JSON の文字列になります。
message = {"MessageId": "m-1", "ReceiptHandle": "h-1", "Body": json.dumps(event)}
print(type(message["Body"]).__name__)
# 受け取る側が、文字列を辞書に戻します。
record = json.loads(message["Body"])["Records"][0]
print(record["s3"]["object"]["key"])
parts = record["s3"]["object"]["key"].split("/")
print(parts)
# key を "orders/tanaka/7.json" に変えて、もう一度実行しましょう。
メッセージは2回届くことがある
キューは、少なくとも1回の配信を約束します。たとえば、ワーカーの処理が遅いと、同じメッセージがもう一度渡されることがあります。
だから、関数は冪等でなければなりません。同じ注文をもう一度書いても、2件目の注文ができてはいけません。条件つきの書き込みが、それを守ります。そのキーに何も入っていないときだけ、項目を置くのです。
よくある間違い
注意
各メッセージは、ちょうど1回だけ届く、と考える人が多いです。届きません。キューは、同じメッセージを何度か渡すことがあります。どのハンドラーも、繰り返しで何も変わらないように書きます。2回目の書き込みが断られたら、それは落ちる理由ではなく、仕事が済んだという印です。
実行: 2回目の書き込みは断られる
table = {}
def put_item_once(table, item):
key = item["customer"]["S"] + "#" + item["order_id"]["N"]
if key in table:
return {"error": {"Code": "ConditionalCheckFailedException",
"HTTPStatusCode": 400}}
table[key] = item
return {}
item = {"customer": {"S": "sato"}, "order_id": {"N": "3"},
"status": {"S": "received"}}
first = put_item_once(table, item)
second = put_item_once(table, item)
print(first)
print(second["error"]["Code"])
print(len(table))
# 2つ目の item の order_id を "4" に変えると、項目は何件になりますか。
仕事が済んでから、メッセージを消す
ワーカーが、メッセージを何件かまとめて受け取ります。1件だけ壊れていて、残りは正常です。
メッセージは、1件ずつ別に扱います。仕事を保存してから、そのメッセージを消します。扱えなかったメッセージは、キューに残り、可視性タイムアウトのあとで戻ってきます。何も失われず、早すぎる削除もありません。
実行: 壊れた1件を含むまとまり
queue = [
{"MessageId": "m-1", "Body": "orders/sato/3.json"},
{"MessageId": "m-2", "Body": "broken"},
{"MessageId": "m-3", "Body": "orders/tanaka/7.json"},
]
done = []
waiting = []
for message in queue:
parts = message["Body"].split("/")
if len(parts) == 3:
done.append(message["MessageId"])
else:
waiting.append(message["MessageId"])
print("消した:", done)
print("残した:", waiting)
# "broken" を "orders/suzuki/9.json" に変えると、残るのは何ですか。