分布式事务

TCC 与 2PC 对比

两种实现方式

补偿事务示例（下单失败回滚）

// Python + 伪代码展示 Saga Orchestration 补偿逻辑

from dataclasses import dataclass
from typing import List, Callable
import logging

@dataclass
class SagaStep:
    name: str
    action: Callable       # 正向操作
    compensate: Callable  # 补偿操作

class SagaOrchestrator:
    def __init__(self, steps: List[SagaStep]):
        self.steps = steps
        self.executed: List[SagaStep] = []

    def execute(self, context: dict) -> bool:
        for step in self.steps:
            try:
                logging.info(f"执行步骤: {step.name}")
                step.action(context)
                self.executed.append(step)
            except Exception as e:
                logging.error(f"步骤 {step.name} 失败: {e}，开始补偿")
                self._compensate(context)
                return False
        return True

    def _compensate(self, context: dict):
        # 逆序执行已成功步骤的补偿操作
        for step in reversed(self.executed):
            try:
                logging.info(f"补偿步骤: {step.name}")
                step.compensate(context)
            except Exception as e:
                # 补偿失败需人工介入或进入死信队列
                logging.critical(f"补偿失败，需人工处理: {step.name}, {e}")

# 使用示例：电商下单 Saga
def create_order_saga(order_data: dict):
    saga = SagaOrchestrator([
        SagaStep(
            name="创建订单",
            action=lambda ctx: order_service.create(ctx),
            compensate=lambda ctx: order_service.cancel(ctx["order_id"])
        ),
        SagaStep(
            name="扣减库存",
            action=lambda ctx: inventory_service.deduct(ctx),
            compensate=lambda ctx: inventory_service.restore(ctx["sku_id"], ctx["qty"])
        ),
        SagaStep(
            name="扣款支付",
            action=lambda ctx: payment_service.charge(ctx),
            compensate=lambda ctx: payment_service.refund(ctx["payment_id"])
        ),
    ])
    return saga.execute(order_data)

-- 1. 建表：outbox 消息表
CREATE TABLE outbox_messages (
    id          BIGINT       PRIMARY KEY AUTO_INCREMENT,
    aggregate_id VARCHAR(64) NOT NULL,  -- 业务主键，如 order_id
    event_type   VARCHAR(128) NOT NULL, -- 事件类型，如 ORDER_CREATED
    payload      JSON        NOT NULL,  -- 事件内容
    status       VARCHAR(16) DEFAULT 'PENDING',  -- PENDING/PUBLISHED/FAILED
    created_at   DATETIME    DEFAULT NOW(),
    published_at DATETIME,
    retry_count  INT         DEFAULT 0
);

-- 2. 业务操作 + 写消息表，同一本地事务
BEGIN;
  -- 业务操作
  INSERT INTO orders (id, user_id, amount, status)
  VALUES (1001, 42, 299.00, 'CREATED');

  -- 同事务写 outbox
  INSERT INTO outbox_messages (aggregate_id, event_type, payload)
  VALUES ('1001', 'ORDER_CREATED',
    '{"orderId":1001,"userId":42,"amount":299.00}');
COMMIT;

-- 3. Outbox Worker (Python 轮询投递)
--    定时执行，发现 PENDING 消息则投递到 Kafka

# Python: Outbox Worker
import time
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')

def outbox_worker():
    while True:
        # 批量取 PENDING 消息（加 FOR UPDATE 防并发）
        msgs = db.query("""
            SELECT * FROM outbox_messages
            WHERE status = 'PENDING' AND retry_count < 5
            ORDER BY id LIMIT 100 FOR UPDATE SKIP LOCKED
        """)
        for msg in msgs:
            try:
                producer.send(
                    topic=msg['event_type'],
                    key=msg['aggregate_id'].encode(),
                    value=msg['payload'].encode()
                )
                db.execute("""
                    UPDATE outbox_messages
                    SET status='PUBLISHED', published_at=NOW()
                    WHERE id=?
                """, [msg['id']])
            except Exception as e:
                db.execute("""
                    UPDATE outbox_messages
                    SET retry_count=retry_count+1,
                        status=IF(retry_count>=4,'FAILED','PENDING')
                    WHERE id=?
                """, [msg['id']])
        time.sleep(1)

# 消费者必须保证幂等（检查是否已处理过该 aggregate_id）