キューの再配信と重複処理の防止
受信・業務更新・完了通知を別の段階として追い、再配送で残高が二重に増えない理由を説明します。
説明
キューから受信した仕事は、処理後に完了を通知します。一定期間内に通知がなければ再配送する方式では、同じ仕事を複数回受け取ることがあります。業務更新と重複検出を一緒に確定し、二重処理を防ぎます。
この演習ではSQLiteで仕事と処理済みIDを保存します。不可視期間10秒、試行上限3回の固定時計モデルを使い、残高更新後に完了通知だけ失われた場合を再現します。最後に毎回失敗する仕事をDLQへ移します。
受信・業務更新・完了通知を別の段階として追い、再配送で残高が二重に増えない理由を説明します。
1仕事と残高を用意する
order-7は残高を100増やす仕事です。初期残高は0、処理済みIDの表は空です。
from jobs import QueueSQLiteを使う固定キューを読み込む
queue = Queue()空のキューと残高を用意する
queue.send仕事を登録する("order-7", 100)
queue.show()配送と処理済みの記録を読むorder-7はqueued、試行回数0です。配送状態と処理済み表は別々に保持されています。
2受信して業務更新する
受信時にreceiptが発行され、10秒間見えなくなります。業務更新は処理済みIDの挿入と同じトランザクションで確定します。
first = queue.receive()受信票付きで仕事を取得する
queue.apply(first)処理済みIDと残高を同じトランザクションで更新する残高は100になりますが、完了通知はまだ送っていません。仕事は受信中のままです。
3可視性タイムアウト中のメッセージは再取得できない
同じ時刻でもう一度receiveします。この固定モデルでは受信中の仕事を返しません。
queue.receive()不可視期間中の取得を試すno_visible_messageです。実際のサービスの配送保証は別途確認が必要で、不可視期間だけで重複処理全般を防げるとは限りません。
4完了通知がないまま期限を迎える
時計を10秒進めます。業務更新は済んでいますが通知がないため、同じIDの仕事が再配送されます。
queue.advance(10)不可視期間の終わりへ時計を進める
second = queue.receive()同じ仕事を再受信する試行2回目で新しいreceipt order-7/2が発行されます。イベントIDと受信票は用途が異なります。
5再配送後も残高を増やさない
処理済み表の主キーorder-7がすでに存在するため、同じ業務更新を繰り返しません。
queue.apply(second)処理済みIDを照合して重複更新を防ぐapplied=False、残高100のままです。同じIDを別の金額へ使い回すことはこの例の契約外であり、実装では内容の一致も検査します。
6古い受信票と現在の受信票
古いreceiptではなく、現在の受信に対応するreceiptで完了通知します。
queue.ack(first)前回の受信票で通知を試す
queue.ack(second)現在の受信票で通知する
queue.show()配送完了と処理済みの両方を確認する古い票はFalse、現在の票はTrueとなり、order-7はdoneになります。処理済み記録は残ります。
7毎回失敗する仕事を再試行する
負の金額を持つbadを登録します。3回受信しても検査に失敗するため、どの回も完了通知を送れません。
queue.send("bad", -1)不正な金額の仕事を登録する
for attempt in range(3):上限まで再配送と失敗を繰り返す
message = queue.receive()
queue.apply(message)
queue.advance(10)invalid_amountが3回続き、時計は40になります。order-7の残高100へ負の値が反映されることはありません。
8上限を超える仕事を隔離する
次の受信時に試行回数を確認し、3回失敗した仕事をDLQへ移します。
queue.receive()試行上限に達した仕事を隔離する
queue.show()通常の完了とDLQを確認するbadはdlq、order-7はdoneです。DLQへ移すことは仕事の成功ではなく、通常の再試行から外して調査対象にする処理です。
まとめ
業務更新後に完了通知が失われると、同じ仕事が再配送されます。
処理済みIDと業務更新を同じトランザクションで確定すると、二重更新を防げます。
DLQは処理成功を意味せず、失敗した仕事を保持します。
操作履歴と解説を振り返れます。
Python 3.15 / SQLite / 固定時計のイベント処理モデル
from jobs import Queue
queue = Queue()
queue.send("order-7", 100)
queue.show()queued=order-7
[('order-7', 0, 'queued')]
processed=[]first = queue.receive()
queue.apply(first)id=order-7 attempt=1 receipt=order-7/1 invisible_until=10 applied=True balance=100
queue.receive()no_visible_message
queue.advance(10)
second = queue.receive()now=10 id=order-7 attempt=2 receipt=order-7/2 invisible_until=20
queue.apply(second)applied=False balance=100
queue.ack(first)
queue.ack(second)
queue.show()ack=False
ack=True
[('order-7', 2, 'done')]
processed=[('order-7',)]queue.send("bad", -1)
for attempt in range(3):
message = queue.receive()
queue.apply(message)
queue.advance(10)queued=bad id=bad attempt=1 receipt=bad/1 invisible_until=20 invalid_amount now=20 id=bad attempt=2 receipt=bad/2 invisible_until=30 invalid_amount now=30 id=bad attempt=3 receipt=bad/3 invisible_until=40 invalid_amount now=40
queue.receive()
queue.show()DLQ=bad
[('bad', 3, 'dlq'), ('order-7', 2, 'done')]
processed=[('order-7',)]演習完了