1
-- 本地消息表
2
CREATE TABLE outbox_messages (
3
    id            BIGINT AUTO_INCREMENT PRIMARY KEY,
4
    topic         VARCHAR(255) NOT NULL COMMENT '消息 Topic',
5
    message_key   VARCHAR(255) NOT NULL COMMENT '消息 Key（用于分区路由和幂等）',
6
    message_value TEXT NOT NULL COMMENT '消息内容（JSON）',
7
    status        ENUM('PENDING', 'SENT', 'FAILED') DEFAULT 'PENDING' COMMENT '发送状态',
8
    retry_count   INT DEFAULT 0 COMMENT '重试次数',
9
    max_retries   INT DEFAULT 5 COMMENT '最大重试次数',
10
    next_retry_at TIMESTAMP NULL COMMENT '下次重试时间（指数退避）',
11
    created_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
12
    updated_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
13

14
    INDEX idx_status_next_retry (status, next_retry_at),
15
    INDEX idx_created (created_at)
16
) ENGINE=InnoDB COMMENT='本地消息表';

1
@Service
2
public class OrderService {
3
    private final OrderRepository orderRepository;
4
    private final OutboxMessageRepository outboxRepository;
5
    private final KafkaTemplate<String, String> kafkaTemplate;
6

7
    @Transactional
8
    public Order createOrder(CreateOrderRequest request) {
9
        // 1. 写入业务数据
10
        Order order = new Order();
11
        order.setId(UUID.randomUUID().toString());
12
        order.setCustomerId(request.getCustomerId());
13
        order.setAmount(request.getAmount());
14
        order.setStatus(OrderStatus.CREATED);
15
        order.setItems(request.getItems());
16
        orderRepository.save(order);
17

18
        // 2. 写入消息表（同一事务，保证原子性）
19
        OutboxMessage message = new OutboxMessage();
20
        message.setTopic("order-events");
21
        message.setMessageKey(order.getId());  // orderId 作为 Key，保证分区有序
22
        message.setMessageValue(JSON.toJSONString(order));
23
        message.setStatus(MessageStatus.PENDING);
24
        message.setRetryCount(0);
25
        message.setMaxRetries(5);
26
        message.setNextRetryAt(LocalDateTime.now());  // 立即可发送
27
        outboxRepository.save(message);
28

29
        return order;
30
    }
31

32
    // 定时任务：扫描待发送消息
33
    @Scheduled(fixedRate = 5000)
34
    public void sendPendingMessages() {
35
        List<OutboxMessage> pending = outboxRepository
36
            .findByStatusAndNextRetryAtBefore(MessageStatus.PENDING, LocalDateTime.now());
37

38
        for (OutboxMessage msg : pending) {
39
            try {
40
                // 同步发送，确保确认后再更新状态
41
                kafkaTemplate.send(msg.getTopic(), msg.getMessageKey(), msg.getMessageValue())
42
                    .get(10, TimeUnit.SECONDS);
43
                msg.setStatus(MessageStatus.SENT);
44
            } catch (Exception e) {
45
                msg.setRetryCount(msg.getRetryCount() + 1);
46
                if (msg.getRetryCount() >= msg.getMaxRetries()) {
47
                    msg.setStatus(MessageStatus.FAILED);
48
                    // 告警：消息发送最终失败
49
                    alertService.alert("消息发送失败: " + msg.getId());
50
                } else {
51
                    // 指数退避：1s, 2s, 4s, 8s, 16s
52
                    long delaySeconds = (long) Math.pow(2, msg.getRetryCount());
53
                    msg.setNextRetryAt(LocalDateTime.now().plusSeconds(delaySeconds));
54
                }
55
            }
56
            outboxRepository.save(msg);
57
        }
58
    }
59

60
    // 定期清理已发送消息（保留 7 天）
61
    @Scheduled(cron = "0 0 3 * * ?")
62
    public void cleanSentMessages() {
63
        outboxRepository.deleteByStatusAndUpdatedAtBefore(
64
            MessageStatus.SENT,
65
            LocalDateTime.now().minusDays(7)
66
        );
67
    }
68
}

1
import json
2
import uuid
3
from datetime import datetime, timedelta
4
from sqlalchemy import create_engine, Column, String, Integer, Enum, DateTime, Text
5
from sqlalchemy.orm import sessionmaker, declarative_base
6
from kafka import KafkaProducer
7

8
Base = declarative_base()
9

10
class OutboxMessage(Base):
11
    __tablename__ = 'outbox_messages'
12
    id = Column(Integer, primary_key=True, autoincrement=True)
13
    topic = Column(String(255), nullable=False)
14
    message_key = Column(String(255), nullable=False)
15
    message_value = Column(Text, nullable=False)
16
    status = Column(Enum('PENDING', 'SENT', 'FAILED'), default='PENDING')
17
    retry_count = Column(Integer, default=0)
18
    next_retry_at = Column(DateTime, default=datetime.now)
19

20
class OrderService:
21
    def __init__(self, db_url: str, kafka_servers: list):
22
        self.engine = create_engine(db_url)
23
        self.Session = sessionmaker(bind=self.engine)
24
        self.producer = KafkaProducer(
25
            bootstrap_servers=kafka_servers,
26
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
27
            key_serializer=lambda k: k.encode('utf-8'),
28
            acks='all',
29
            retries=3
30
        )
31

32
    def create_order(self, customer_id: str, amount: float, items: list):
33
        session = self.Session()
34
        try:
35
            order_id = str(uuid.uuid4())
36
            # 1. 写入业务数据
37
            session.execute(
38
                "INSERT INTO orders (id, customer_id, amount, status) "
39
                "VALUES (:id, :cid, :amt, 'CREATED')",
40
                {"id": order_id, "cid": customer_id, "amt": amount}
41
            )
42
            # 2. 写入消息表（同一事务）
43
            session.execute(
44
                "INSERT INTO outbox_messages (topic, message_key, message_value, status) "
45
                "VALUES ('order-events', :key, :value, 'PENDING')",
46
                {
47
                    "key": order_id,
48
                    "value": json.dumps({
49
                        "orderId": order_id, "customerId": customer_id,
50
                        "amount": amount, "items": items,
51
                        "eventType": "ORDER_CREATED"
52
                    })
53
                }
54
            )
55
            session.commit()
56
            return order_id
57
        except Exception:
58
            session.rollback()
59
            raise
60

61
    def send_pending_messages(self):
62
        session = self.Session()
63
        messages = session.execute(
64
            "SELECT * FROM outbox_messages "
65
            "WHERE status = 'PENDING' AND next_retry_at <= :now "
66
            "LIMIT 100",
67
            {"now": datetime.now()}
68
        ).fetchall()
69

70
        for msg in messages:
71
            try:
72
                future = self.producer.send(
73
                    msg.topic, key=msg.message_key, value=json.loads(msg.message_value)
74
                )
75
                future.get(timeout=10)  # 同步等待确认
76
                session.execute(
77
                    "UPDATE outbox_messages SET status = 'SENT' WHERE id = :id",
78
                    {"id": msg.id}
79
                )
80
            except Exception:
81
                new_retry = msg.retry_count + 1
82
                if new_retry >= 5:
83
                    session.execute(
84
                        "UPDATE outbox_messages SET status = 'FAILED' WHERE id = :id",
85
                        {"id": msg.id}
86
                    )
87
                else:
88
                    delay = 2 ** new_retry
89
                    session.execute(
90
                        "UPDATE outbox_messages SET retry_count = :rc, "
91
                        "next_retry_at = :nrt WHERE id = :id",
92
                        {"rc": new_retry, "nrt": datetime.now() + timedelta(seconds=delay), "id": msg.id}
93
                    )
94
        session.commit()

1
// Debezium Connector 配置
2
{
3
  "name": "order-outbox-connector",
4
  "config": {
5
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
6
    "database.hostname": "mysql",
7
    "database.port": "3306",
8
    "database.user": "debezium",
9
    "database.password": "password",
10
    "database.server.id": "184054",
11
    "database.server.name": "order_service",
12
    "database.include.list": "order_db",
13
    "table.include.list": "order_db.outbox_messages",
14
    "database.history.kafka.bootstrap.servers": "kafka:9092",
15
    "database.history.kafka.topic": "schema-changes.order",
16
    "transforms": "outbox",
17
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
18
    "transforms.outbox.route.topic.replacement": "${routedByValue}",
19
    "transforms.outbox.table.field.event.id": "id",
20
    "transforms.outbox.table.field.event.key": "message_key",
21
    "transforms.outbox.table.field.event.type": "event_type",
22
    "transforms.outbox.table.field.event.payload": "message_value"
23
  }
24
}

1
// 应用端：只需写入 Outbox 表
2
@Service
3
public class OrderService {
4
    @Transactional
5
    public void createOrder(Order order) {
6
        // 1. 写入业务数据
7
        orderRepository.save(order);
8

9
        // 2. 写入 Outbox 表（同一事务）
10
        OutboxEvent event = new OutboxEvent();
11
        event.setAggregateType("Order");
12
        event.setAggregateId(order.getId());
13
        event.setEventType("ORDER_CREATED");
14
        event.setPayload(JSON.toJSONString(order));
15
        outboxRepository.save(event);
16

17
        // Debezium 会自动捕获 Outbox 表的变更并发送到 Kafka
18
        // 无需手动发送消息！
19
    }
20
}

1
-- 标准 Outbox 表（Debezium 推荐格式）
2
CREATE TABLE outbox_events (
3
    id            VARCHAR(255) PRIMARY KEY,     -- 事件 ID（UUID）
4
    aggregatetype VARCHAR(255) NOT NULL,        -- 聚合类型（如 Order）
5
    aggregateid   VARCHAR(255) NOT NULL,        -- 聚合 ID（如订单 ID）
6
    eventtype     VARCHAR(255) NOT NULL,        -- 事件类型（如 ORDER_CREATED）
7
    payload       JSON NOT NULL,                -- 事件负载
8
    created_at    TIMESTAMP DEFAULT NOW(),      -- 创建时间
9

10
    INDEX idx_aggregate (aggregatetype, aggregateid),
11
    INDEX idx_created (created_at)
12
);
13

14
-- 清理策略：定期删除已处理的事件
15
-- Debezium 读取后可以安全删除（CDC 已捕获）
16
DELETE FROM outbox_events WHERE created_at < DATE_SUB(NOW(), INTERVAL 7 DAY);

1
{
2
  "name": "order-cdc-connector",
3
  "config": {
4
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
5
    "database.hostname": "mysql",
6
    "database.port": "3306",
7
    "database.user": "debezium",
8
    "database.password": "password",
9
    "database.server.id": "184054",
10
    "database.server.name": "order_service",
11
    "database.include.list": "order_db",
12
    "table.include.list": "order_db.orders,order_db.order_items",
13
    "snapshot.mode": "schema_only",
14
    "database.history.kafka.bootstrap.servers": "kafka:9092",
15
    "database.history.kafka.topic": "schema-changes.order"
16
  }
17
}

1
# 注册 Debezium Connector
2
curl -X POST -H "Content-Type: application/json" \
3
    --data @debezium-order.json \
4
    http://localhost:8083/connectors
5

6
# 查看 Connector 状态
7
curl http://localhost:8083/connectors/order-cdc-connector/status

1
// 消费 Debezium 变更事件
2
@KafkaListener(topics = "order_service.order_db.orders")
3
public void handleOrderChange(ConsumerRecord<String, String> record) {
4
    JSONObject event = JSON.parseObject(record.value());
5
    JSONObject payload = event.getJSONObject("payload");
6

7
    String operation = payload.getString("op");
8
    JSONObject before = payload.getJSONObject("before");
9
    JSONObject after = payload.getJSONObject("after");
10

11
    switch (operation) {
12
        case "c" -> {  // Create
13
            syncToElasticsearch(after);
14
            syncToRedis(after);
15
        }
16
        case "u" -> {  // Update
17
            syncToElasticsearch(after);
18
            invalidateRedisCache(after);
19
        }
20
        case "d" -> {  // Delete
21
            deleteFromElasticsearch(before.getString("id"));
22
            deleteFromRedis(before.getString("id"));
23
        }
24
    }
25
}

1
// Saga 编排器：定义事务步骤和补偿操作
2
@Service
3
public class OrderSagaOrchestrator {
4
    private final KafkaTemplate<String, String> kafkaTemplate;
5

6
    public void executeOrderSaga(Order order) {
7
        // 步骤 1：创建订单（已在 OrderService 中完成，Outbox 已发送）
8
        // 等待库存服务消费 ORDER_CREATED 事件
9

10
        // Saga 状态机
11
        sagaStateRepository.save(new SagaState(order.getId(), "INVENTORY_DEDUCTING"));
12
    }
13

14
    // 库存扣减成功
15
    @KafkaListener(topics = "inventory-events", groupId = "saga-orchestrator")
16
    public void onInventoryEvent(ConsumerRecord<String, String> record) {
17
        InventoryEvent event = JSON.parseObject(record.value(), InventoryEvent.class);
18
        SagaState saga = sagaStateRepository.findByOrderId(event.getOrderId());
19

20
        switch (event.getEventType()) {
21
            case "INVENTORY_DEDUCTED" -> {
22
                // 库存扣减成功 → 触发支付
23
                saga.setState("PAYMENT_PROCESSING");
24
                sendPaymentCommand(event.getOrderId());
25
            }
26
            case "INVENTORY_DEDUCT_FAILED" -> {
27
                // 库存扣减失败 → 取消订单
28
                saga.setState("COMPENSATING");
29
                sendCancelOrderCommand(event.getOrderId());
30
            }
31
        }
32
        sagaStateRepository.save(saga);
33
    }
34

35
    // 支付结果
36
    @KafkaListener(topics = "payment-events", groupId = "saga-orchestrator")
37
    public void onPaymentEvent(ConsumerRecord<String, String> record) {
38
        PaymentEvent event = JSON.parseObject(record.value(), PaymentEvent.class);
39
        SagaState saga = sagaStateRepository.findByOrderId(event.getOrderId());
40

41
        switch (event.getEventType()) {
42
            case "PAYMENT_COMPLETED" -> {
43
                // 支付成功 → 确认订单
44
                saga.setState("COMPLETED");
45
                sendConfirmOrderCommand(event.getOrderId());
46
            }
47
            case "PAYMENT_FAILED" -> {
48
                // 支付失败 → 补偿：恢复库存 + 取消订单
49
                saga.setState("COMPENSATING");
50
                sendRestoreInventoryCommand(event.getOrderId());
51
                sendCancelOrderCommand(event.getOrderId());
52
            }
53
        }
54
        sagaStateRepository.save(saga);
55
    }
56

57
    private void sendPaymentCommand(String orderId) {
58
        // 通过 Outbox 发送支付命令
59
        outboxService.send("payment-commands", orderId,
60
            JSON.toJSONString(Map.of("orderId", orderId, "command", "PROCESS_PAYMENT")));
61
    }
62
}

1
// 使用 RocketMQ 事务消息替代本地消息表
2
TransactionMQProducer producer = new TransactionMQProducer("order-producer-group");
3
producer.setTransactionListener(new TransactionListener() {
4
    @Override
5
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
6
        try {
7
            orderService.createOrder(parseOrder(msg));
8
            return LocalTransactionState.COMMIT_MESSAGE;
9
        } catch (Exception e) {
10
            return LocalTransactionState.ROLLBACK_MESSAGE;
11
        }
12
    }
13

14
    @Override
15
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
16
        Order order = orderService.getById(parseOrderId(msg));
17
        return order != null ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
18
    }
19
});