事件驱动架构深度实战 — Event Sourcing / CQRS / Saga / Outbox

# consumer_with_inbox.py
import json, sqlite3
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'orders.orderpaid',
    bootstrap_servers='kafka:9092',
    enable_auto_commit=False,           # 关键:关闭自动提交
    auto_offset_reset='earliest',
    group_id='read-model-builder',
)

db = sqlite3.connect('read_model.db')

for msg in consumer:
    payload = json.loads(msg.value)
    event_id = payload['id']

    try:
        # 1. 幂等表插入(冲突即说明已处理过)
        db.execute(
            "INSERT INTO inbox(event_id, payload) VALUES(?, ?)",
            (event_id, json.dumps(payload))
        )
        # 2. 真正的业务处理
        apply_to_read_model(db, payload)
        # 3. 提交位点
        db.commit()
        consumer.commit()
    except sqlite3.IntegrityError:
        # event_id 重复,直接跳过
        db.rollback()
        consumer.commit()
    except Exception as e:
        # 处理失败,回滚,让 Kafka 重投
        db.rollback()
        raise

# es_core.py
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Dict, Any, Callable
from uuid import uuid4

# ────────────────── 事件基类 ──────────────────
@dataclass(frozen=True)
class Event:
    event_id: str = field(default_factory=lambda: str(uuid4()))
    occurred_at: datetime = field(default_factory=datetime.utcnow)
    version: int = 1

# ────────────────── 具体事件 ──────────────────
@dataclass(frozen=True)
class OrderCreated(Event):
    order_id: str = ''
    customer_id: str = ''
    items: tuple = ()
    total_amount: int = 0

@dataclass(frozen=True)
class OrderPaid(Event):
    order_id: str = ''
    paid_amount: int = 0
    payment_tx: str = ''

@dataclass(frozen=True)
class OrderShipped(Event):
    order_id: str = ''
    tracking_no: str = ''
    carrier: str = ''

@dataclass(frozen=True)
class OrderCancelled(Event):
    order_id: str = ''
    reason: str = ''

# ────────────────── 命令基类 ──────────────────
@dataclass(frozen=True)
class Command:
    command_id: str = field(default_factory=lambda: str(uuid4()))

@dataclass(frozen=True)
class CreateOrder(Command):
    customer_id: str = ''
    items: tuple = ()

@dataclass(frozen=True)
class PayOrder(Command):
    order_id: str = ''
    amount: int = 0
    payment_tx: str = ''

@dataclass(frozen=True)
class ShipOrder(Command):
    order_id: str = ''
    tracking_no: str = ''

# ────────────────── Aggregate(订单聚合根) ──────────────────
class OrderAggregate:
    STATUS = ('PENDING', 'PAID', 'SHIPPED', 'CANCELLED')

    def __init__(self, order_id: str):
        self.order_id = order_id
        self.status = 'PENDING'
        self.items = ()
        self.total_amount = 0
        self.version = 0             # 乐观锁
        self._changes: List[Event] = []

    # ── 命令处理入口(决定产生哪些事件) ──
    def create(self, cmd: CreateOrder) -> OrderCreated:
        if self.status != 'PENDING':
            raise ValueError(f'订单 {self.order_id} 已存在,不能重复创建')
        if not cmd.items:
            raise ValueError('订单至少 1 件商品')
        event = OrderCreated(
            order_id=self.order_id,
            customer_id=cmd.customer_id,
            items=cmd.items,
            total_amount=sum(i['price'] * i['qty'] for i in cmd.items),
        )
        self._apply(event)
        return event

    def pay(self, cmd: PayOrder) -> OrderPaid:
        if self.status != 'PENDING':
            raise ValueError(f'订单 {self.order_id} 状态为 {self.status},无法支付')
        if cmd.amount != self.total_amount:
            raise ValueError(f'支付金额 {cmd.amount} != 订单金额 {self.total_amount}')
        event = OrderPaid(
            order_id=self.order_id,
            paid_amount=cmd.amount,
            payment_tx=cmd.payment_tx,
        )
        self._apply(event)
        return event

    def ship(self, cmd: ShipOrder) -> OrderShipped:
        if self.status != 'PAID':
            raise ValueError(f'订单 {self.order_id} 状态为 {self.status},无法发货')
        event = OrderShipped(
            order_id=self.order_id,
            tracking_no=cmd.tracking_no,
            carrier='顺丰',
        )
        self._apply(event)
        return event

    # ── 事件应用入口(改变状态) ──
    def _apply(self, event: Event):
        if isinstance(event, OrderCreated):
            self.status = 'PENDING'
            self.items = event.items
            self.total_amount = event.total_amount
        elif isinstance(event, OrderPaid):
            self.status = 'PAID'
        elif isinstance(event, OrderShipped):
            self.status = 'SHIPPED'
        elif isinstance(event, OrderCancelled):
            self.status = 'CANCELLED'
        self.version += 1
        self._changes.append(event)

    # ── 从事件流 replay ──
    @classmethod
    def load_from_history(cls, order_id: str, history: List[Event]) -> 'OrderAggregate':
        agg = cls(order_id)
        for e in history:
            agg._apply(e)
        agg._changes = []              # replay 不算"新产生"的事件
        return agg

    @property
    def uncommitted_changes(self) -> List[Event]:
        return list(self._changes)

    def mark_changes_committed(self):
        self._changes.clear()

# repository.py
class EventStore:
    def __init__(self, conn):
        self.conn = conn
        self.handlers: Dict[type, Callable[[Any], None]] = {}

    def append(self, aggregate_id: str, expected_version: int, events: List[Event]):
        with self.conn:
            for e in events:
                self.conn.execute(
                    'INSERT INTO events(stream_id, version, type, payload, occurred_at) '
                    'VALUES (?, ?, ?, ?, ?)',
                    (aggregate_id, expected_version := expected_version + 1,
                     type(e).__name__, e_to_json(e), e.occurred_at)
                )

    def load(self, aggregate_id: str, aggregate_cls) -> Any:
        rows = self.conn.execute(
            'SELECT type, payload FROM events WHERE stream_id=? ORDER BY version',
            (aggregate_id,)
        ).fetchall()
        history = [json_to_event(r['type'], r['payload']) for r in rows]
        return aggregate_cls.load_from_history(aggregate_id, history)

def load_with_snapshot(aggregate_id, conn, aggregate_cls):
    snap = conn.execute(
        'SELECT version, payload FROM snapshots WHERE stream_id=? ORDER BY version DESC LIMIT 1',
        (aggregate_id,)
    ).fetchone()
    if snap:
        state = deserialize(payload=snap['payload'])
        agg = aggregate_cls.from_snapshot(aggregate_id, state)
        after = snap['version']
    else:
        agg = aggregate_cls(aggregate_id)
        after = 0
    rows = conn.execute(
        'SELECT type, payload FROM events WHERE stream_id=? AND version > ? ORDER BY version',
        (aggregate_id, after)
    ).fetchall()
    for r in rows:
        agg._apply(json_to_event(r['type'], r['payload']))
    return agg

# upcasting.py
def upcast(event_type: str, payload: dict) -> dict:
    """从 v1 升级到 v2 的兼容层"""
    if event_type == 'OrderPaid':
        # v1: { paidAmount }; v2 增加 paidCurrency
        if 'paidCurrency' not in payload:
            payload['paidCurrency'] = 'CNY'    # 假定当时统一人民币
    if event_type == 'OrderCreated':
        # v1: items 是 dict, v2 是 dict + taxRate
        for it in payload.get('items', []):
            it.setdefault('taxRate', 0.13)
    return payload

def json_to_event(event_type: str, payload: dict) -> Event:
    payload = upcast(event_type, payload)    # 转换是只读的
    cls = EVENT_REGISTRY[event_type]
    return cls(**payload)

# write_side.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import uuid4

app = FastAPI()
event_store = EventStore(get_db_conn())    # 见 4.2
bus = KafkaProducer(bootstrap_servers='kafka:9092')

class CreateOrderCmd(BaseModel):
    customerId: str
    items: list

@app.post('/orders')
def create_order(cmd: CreateOrderCmd):
    order_id = str(uuid4())
    aggregate = OrderAggregate(order_id)
    try:
        event = aggregate.create(CreateOrder(
            customer_id=cmd.customerId,
            items=tuple(cmd.items),
        ))
    except ValueError as e:
        raise HTTPException(400, str(e))

    # ── 写端只做两件:存事件 + 投事件 ──
    event_store.append(order_id, 0, [event])

    # 投给 Kafka(Inbox/Outbox 见第 7 节;这里只是骨架)
    bus.send('orders.created', event_envelope(event))
    return {'orderId': order_id}

# projection_order_list.py
# 单线程消费 orders.* 事件,维护一个订单列表读模型
def project(event: Event, read_db):
    if isinstance(event, OrderCreated):
        read_db.execute("""
            INSERT INTO order_list_view(order_id, customer_id, total_amount, status, created_at)
            VALUES (?, ?, ?, 'PENDING', ?)
        """, (event.order_id, event.customer_id, event.total_amount, event.occurred_at))
    elif isinstance(event, OrderPaid):
        read_db.execute(
            "UPDATE order_list_view SET status='PAID', paid_at=? WHERE order_id=?",
            (event.occurred_at, event.order_id))
    elif isinstance(event, OrderShipped):
        read_db.execute(
            "UPDATE order_list_view SET status='SHIPPED', tracking_no=? WHERE order_id=?",
            (event.tracking_no, event.order_id))
    elif isinstance(event, OrderCancelled):
        read_db.execute(
            "UPDATE order_list_view SET status='CANCELLED' WHERE order_id=?",
            (event.order_id,))

# projection_stats.py
def project_to_stats(event, redis_client):
    pipe = redis_client.pipeline()
    if isinstance(event, OrderCreated):
        pipe.incr('stats:orders:created:total')
    elif isinstance(event, OrderPaid):
        pipe.incrbyfloat('stats:gmv:total', event.paid_amount / 100)
        pipe.incr('stats:orders:paid:total')
    pipe.execute()

# saga_orchestrator.py
from temporalio import workflow, activity

@activity.defn
async def reserve_inventory(items): ...
@activity.defn
async def charge_payment(order_id, amount): ...
@activity.defn
async def release_inventory(items): ...
@activity.defn
async def refund_payment(order_id): ...

@workflow.defn
class OrderSaga:
    @workflow.run
    async def run(self, order):
        try:
            await workflow.execute_activity(
                reserve_inventory, order.items,
                start_to_close_timeout=timedelta(seconds=10))
            try:
                await workflow.execute_activity(
                    charge_payment, (order.id, order.total),
                    start_to_close_timeout=timedelta(seconds=30))
            except Exception:
                await workflow.execute_activity(release_inventory, order.items)
                raise
        except Exception:
            # 任何更上层失败:补偿
            await workflow.execute_activity(refund_payment, order.id)
            raise

db.execute("UPDATE orders SET status='PAID' WHERE id=?", (order_id,))
kafka.send('orders.paid', payload)   # 1)broker 宕
db.commit()                           # 2)commit 成功 + 提交失败 = 不一致

# order_with_outbox.py
def create_order_with_outbox(cmd):
    with db.begin() as tx:
        event = OrderCreated(...)
        # 业务表
        tx.execute('INSERT INTO orders(id, status) VALUES (?, ?)',
                   (event.order_id, 'PENDING'))
        # outbox 表(同事务)
        tx.execute("""
            INSERT INTO outbox(aggregate_type, aggregate_id, event_type, payload)
            VALUES ('Order', ?, 'OrderCreated', ?::jsonb)
        """, (event.order_id, event.to_json()))
        # ... commit 后,事件一定写入 outbox(事务原子)

# consumer_with_inbox.py
def process(event_envelope):
    with db.begin() as tx:
        # 1. 幂等记录(主键冲突 = 已消费)
        tx.execute("""
            INSERT INTO inbox(event_id) VALUES (?::text)
            ON CONFLICT (event_id) DO NOTHING
        """, (event_envelope.id,))
        # 如果是 UPDATE 0 行,直接跳过
        if tx.rowcount == 0:
            return
        # 2. 真正的业务处理(可能在 Saga 内)
        apply_to_projection(tx, event_envelope)
        # 3. 标记处理完成
        tx.execute('UPDATE inbox SET processed_at=now() WHERE event_id=?', (event_envelope.id,))

# main.py —— 写端 + 事务性 Outbox
import json
from datetime import datetime
from confluent_kafka import Producer
import psycopg

CONN = psycopg.connect('postgresql://app:secret@postgres:5432/orders')

def handle_pay_order(order_id: str, amount: int, tx_no: str):
    with CONN.transaction() as tx:
        cur = tx.cursor()
        # ── 业务校验 ──
        cur.execute('SELECT status, total FROM orders WHERE id=%s FOR UPDATE', (order_id,))
        row = cur.fetchone()
        if not row or row[0] != 'PENDING' or row[1] != amount:
            raise ValueError('invalid state')
        cur.execute("UPDATE orders SET status='PAID', paid_at=now() WHERE id=%s", (order_id,))
        # ── Outbox 写同事务 ──
        cur.execute("""
            INSERT INTO outbox(aggregate_type, aggregate_id, event_type, payload)
            VALUES (%s, %s, %s, %s::jsonb)
        """, ('Order', order_id, 'OrderPaid',
              json.dumps({'orderId': order_id, 'amount': amount, 'txNo': tx_no},
                         default=str)))
        # ── commit ──
    # 此处事务已提交,Outbox 消息一定持久化;Debezium 会异步取走 → Kafka

# inventory_consumer.py —— Debezium 把 Outbox → Kafka,这边订阅
from confluent_kafka import Consumer
import psycopg

INVENTORY_DB = psycopg.connect('postgresql://app:secret@postgres:5432/inventory')

def run():
    c = Consumer({
        'bootstrap.servers': 'kafka:9092',
        'group.id': 'inventory-projector',
        'enable.auto.commit': False,
        'auto.offset.reset': 'earliest',
    })
    c.subscribe(['Order.aggregate'])
    while True:
        msg = c.poll(1)
        if not msg:
            continue
        envelope = json.loads(msg.value())
        event_type = envelope['type']
        event_id = envelope['id']
        payload = envelope['data']
        with INVENTORY_DB.transaction() as tx:
            cur = tx.cursor()
            # 幂等
            cur.execute('INSERT INTO inbox(event_id) VALUES (%s) ON CONFLICT DO NOTHING', (event_id,))
            if cur.rowcount == 0:
                c.commit(msg)
                continue
            if event_type == 'OrderPaid':
                cur.execute(
                    'UPDATE stock SET qty = qty - 1 WHERE sku=%s',
                    (payload['items'][0]['sku'],))
            elif event_type == 'OrderCancelled':
                cur.execute(
                    'UPDATE stock SET qty = qty + 1 WHERE sku=%s',
                    (payload['items'][0]['sku'],))
        c.commit(msg)

# account_aggregate.py
class AccountAggregate:
    def __init__(self, account_id):
        self.account_id = account_id
        self.balance = 0
        self.owner = ''
        self.frozen = False
        self.version = 0
        self._changes = []

    def _apply(self, e):
        if isinstance(e, AccountOpened):
            self.balance, self.owner = 0, e.owner
        elif isinstance(e, MoneyDeposited):
            self.balance += e.amount
        elif isinstance(e, MoneyWithdrawn):
            if self.balance < e.amount:
                raise ValueError('余额不足')
            self.balance -= e.amount
        elif isinstance(e, AccountFrozen):
            self.frozen = True
        self.version += 1
        self._changes.append(e)

    def deposit(self, amount):
        self._apply(MoneyDeposited(amount=amount))
    def withdraw(self, amount):
        self._apply(MoneyWithdrawn(amount=amount))

    @classmethod
    def load_from_history(cls, aid, events):
        agg = cls(aid)
        for e in events:
            agg._apply(e)
        agg._changes = []
        return agg

# replay_at.py
def balance_at(account_id, at_time):
    events = store.load_until(account_id, at_time)
    agg = AccountAggregate.load_from_history(account_id, events)
    return agg.balance

print(balance_at('ACC001', '2026-07-01 12:00:00'))   # 比如 12500
print(balance_at('ACC001', '2026-07-07 09:00:00'))   # 比如 18000