106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
"""发件箱（Outbox）模式：业务数据和“待发送消息”写在同一个数据库事务里，由后台程序投递。
|
|
|
|
只用标准库（sqlite3）演示，直接运行：python outbox_demo.py
|
|
真实项目里把 sqlite 换成 MySQL / PostgreSQL，把 FakeBroker 换成 RabbitMQ / Kafka 客户端。
|
|
"""
|
|
import json
|
|
import sqlite3
|
|
|
|
|
|
class FakeBroker:
|
|
"""模拟消息队列：fail_times 次之内发送失败，用来演示重试。"""
|
|
|
|
def __init__(self, fail_times: int = 0) -> None:
|
|
self.sent: list[dict] = []
|
|
self.fail_times = fail_times
|
|
|
|
def publish(self, topic: str, msg: dict) -> None:
|
|
if self.fail_times > 0:
|
|
self.fail_times -= 1
|
|
raise ConnectionError("broker 暂时不可用")
|
|
self.sent.append({"topic": topic, **msg})
|
|
|
|
|
|
def init(db: sqlite3.Connection) -> None:
|
|
db.executescript("""
|
|
CREATE TABLE orders (id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, amount INTEGER NOT NULL);
|
|
CREATE TABLE outbox (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
topic TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
sent_at TEXT, -- NULL 表示还没发送
|
|
attempts INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
""")
|
|
|
|
|
|
def create_order(db: sqlite3.Connection, user_id: int, amount: int, fail_after_insert: bool = False) -> int:
|
|
with db: # 一个事务：订单和消息要么都写入，要么都不写入
|
|
cur = db.execute("INSERT INTO orders (user_id, amount) VALUES (?, ?)", (user_id, amount))
|
|
order_id = cur.lastrowid
|
|
db.execute("INSERT INTO outbox (topic, payload) VALUES (?, ?)",
|
|
("order.created", json.dumps({"order_id": order_id, "user_id": user_id, "amount": amount})))
|
|
if fail_after_insert:
|
|
raise RuntimeError("模拟业务异常：事务回滚")
|
|
return order_id
|
|
|
|
|
|
def relay(db: sqlite3.Connection, broker: FakeBroker, batch: int = 100) -> int:
|
|
"""后台投递：取未发送的消息，发送成功后标记。可能重复发送（至少一次），消费者要幂等。"""
|
|
rows = db.execute("SELECT id, topic, payload FROM outbox WHERE sent_at IS NULL ORDER BY id LIMIT ?",
|
|
(batch,)).fetchall()
|
|
sent = 0
|
|
for mid, topic, payload in rows:
|
|
try:
|
|
broker.publish(topic, {"message_id": mid, **json.loads(payload)})
|
|
except ConnectionError:
|
|
with db:
|
|
db.execute("UPDATE outbox SET attempts = attempts + 1 WHERE id = ?", (mid,))
|
|
break # 保持顺序：前面的没发出去，后面的先不发
|
|
with db:
|
|
db.execute("UPDATE outbox SET sent_at = datetime('now') WHERE id = ?", (mid,))
|
|
sent += 1
|
|
return sent
|
|
|
|
|
|
class Consumer:
|
|
"""幂等消费者：按 message_id 去重（真实项目里用数据库唯一约束记录已处理的消息）。"""
|
|
|
|
def __init__(self) -> None:
|
|
self.processed: set[int] = set()
|
|
self.points: dict[int, int] = {}
|
|
|
|
def handle(self, msg: dict) -> None:
|
|
if msg["message_id"] in self.processed:
|
|
return
|
|
self.points[msg["user_id"]] = self.points.get(msg["user_id"], 0) + msg["amount"] // 100
|
|
self.processed.add(msg["message_id"])
|
|
|
|
|
|
def main() -> None:
|
|
db = sqlite3.connect(":memory:")
|
|
init(db)
|
|
create_order(db, user_id=1, amount=19900)
|
|
try:
|
|
create_order(db, user_id=2, amount=500, fail_after_insert=True)
|
|
except RuntimeError:
|
|
pass
|
|
create_order(db, user_id=1, amount=10000)
|
|
assert db.execute("SELECT count(*) FROM orders").fetchone()[0] == 2
|
|
assert db.execute("SELECT count(*) FROM outbox").fetchone()[0] == 2 # 回滚的订单没有留下消息
|
|
|
|
broker = FakeBroker(fail_times=1)
|
|
assert relay(db, broker) == 0 # 第一次 broker 不可用
|
|
assert relay(db, broker) == 2 # 重试成功
|
|
assert relay(db, broker) == 0 # 没有待发送的了
|
|
|
|
consumer = Consumer()
|
|
for m in broker.sent + broker.sent[:1]: # 模拟第一条消息被重复投递
|
|
consumer.handle(m)
|
|
assert consumer.points == {1: 199 + 100}, consumer.points
|
|
print("发件箱：订单和消息同事务写入；投递失败可重试；消费者幂等，重复消息不会重复加积分")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|