Eine Bestelldatei, drei Dienste
Eine Kundin lädt die Bestelldatei einer Bäckerei in einen S3-Bucket hoch. Kurz darauf steht die Bestellung als Zeile in einer DynamoDB-Tabelle. Dazwischen hat niemand einen Knopf gedrückt.
Drei kleine Schritte haben das getan: ein Event, eine Warteschlange und eine Funktion. Diese Lektion folgt einem Upload durch alle drei und zeigt, was passiert, wenn ein Schritt scheitert oder zweimal läuft.
Am Ende dieser Lektion
Notiz
Ich kann einem Upload durch eine Warteschlange und eine Funktion bis in eine Tabelle folgen.
Schlüsselwörter
| Begriff | Was er bedeutet | Beispiel |
|---|---|---|
| Pipeline | Schritte, die Arbeit nacheinander weiterreichen | |
| Event | eine Notiz, dass etwas geschehen ist | |
| Nachricht | ein Stück Text, das in einer Warteschlange wartet | |
| Item | eine gespeicherte Zeile in einer DynamoDB-Tabelle | |
| Idempotent | zweimal ausgeführt mit demselben Ergebnis | ein Item nur ablegen, wenn es neu ist |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Jeder Schritt reicht Text weiter
Der Upload lässt S3 ein Event senden. Eine Regel legt dieses Event als Nachricht in eine SQS-Warteschlange. Eine Funktion liest die Nachricht und schreibt ein Item in eine Tabelle.
Die Warteschlange trägt Text, keine Dictionaries. Darum wird das Event zu einem JSON-String, und die Funktion macht daraus wieder ein Dictionary. Dieses Event ist die dokumentierte Form, keines, das ein Stack für Sie erzeugt hat.
Run: aus einem Event wird ein Nachrichtentext
import json
# Die dokumentierte Form eines S3-Events, gekürzt auf das, was wir lesen.
event = {"Records": [{
"eventName": "ObjectCreated:Put",
"s3": {"bucket": {"name": "baeckerei-uploads"},
"object": {"key": "bestellungen/anna/3.json"}}}]}
# Die Warteschlange trägt Text, also wird das Event ein JSON-String.
nachricht = {"MessageId": "m-1", "ReceiptHandle": "h-1", "Body": json.dumps(event)}
print(type(nachricht["Body"]).__name__)
# Der Worker macht aus dem Text wieder ein Dictionary.
eintrag = json.loads(nachricht["Body"])["Records"][0]
print(eintrag["s3"]["object"]["key"])
teile = eintrag["s3"]["object"]["key"].split("/")
print(teile)
# Ändern Sie den Schlüssel in "bestellungen/ben/7.json" und führen Sie es noch einmal aus.
Eine Nachricht kann zweimal kommen
Eine Warteschlange verspricht mindestens einmal. Hin und wieder liefert sie dieselbe Nachricht erneut, etwa wenn ein Worker zu langsam fertig war.
Darum muss die Funktion idempotent sein. Eine Bestellung noch einmal zu schreiben darf keine zweite Bestellung ergeben. Ein bedingtes Schreiben leistet das: Das Item wird nur abgelegt, wenn unter dem Schlüssel noch nichts steht.
Ein häufiger Fehler
Warnung
Viele glauben, jede Nachricht komme genau einmal an. Das stimmt nicht: Eine Warteschlange kann eine Nachricht mehrfach liefern. Schreiben Sie jeden Handler so, dass eine Wiederholung nichts ändert. Das zweite Schreiben ist kein Grund abzustürzen, es heißt nur: Die Arbeit ist schon getan.
Run: das zweite Schreiben wird abgelehnt
tabelle = {}
def item_einmal_ablegen(tabelle, item):
schluessel = item["kunde"]["S"] + "#" + item["bestell_id"]["N"]
if schluessel in tabelle:
return {"error": {"Code": "ConditionalCheckFailedException",
"HTTPStatusCode": 400}}
tabelle[schluessel] = item
return {}
item = {"kunde": {"S": "anna"}, "bestell_id": {"N": "3"},
"status": {"S": "erhalten"}}
erstes = item_einmal_ablegen(tabelle, item)
zweites = item_einmal_ablegen(tabelle, item)
print(erstes)
print(zweites["error"]["Code"])
print(len(tabelle))
# Ändern Sie die bestell_id des zweiten Aufrufs auf "4". Wie viele Items sind es dann?
Eine Nachricht löschen, wenn ihre Arbeit getan ist
Ein Worker holt einen Stapel Nachrichten ab. Eine davon ist kaputt. Die anderen sind in Ordnung.
Behandeln Sie jede Nachricht für sich. Nachdem die Arbeit gespeichert ist, löschen Sie diese Nachricht. Eine Nachricht, die Sie nicht behandeln konnten, bleibt in der Warteschlange und kehrt nach ihrem Sichtbarkeits-Timeout zurück. Nichts geht verloren, und nichts wird zu früh gelöscht.
Run: ein Stapel mit einer schlechten Nachricht
schlange = [
{"MessageId": "m-1", "Body": "bestellungen/anna/3.json"},
{"MessageId": "m-2", "Body": "kaputt"},
{"MessageId": "m-3", "Body": "bestellungen/ben/7.json"},
]
geloescht = []
uebrig = []
for nachricht in schlange:
teile = nachricht["Body"].split("/")
if len(teile) == 3:
geloescht.append(nachricht["MessageId"])
else:
uebrig.append(nachricht["MessageId"])
print("gelöscht:", geloescht)
print("übrig:", uebrig)
# Ändern Sie "kaputt" in "bestellungen/cem/9.json". Was bleibt jetzt übrig?