Kafka 消息可靠性与幂等处理：Transactional Outbox 实战

订单服务经常需要在同一次业务操作后更新数据库并发出 Kafka 事件。先提交数据库、再发消息，进程可能在两步之间崩溃；先发消息、再提交数据库，消费者可能读到并不存在的订单。除非数据库和消息代理共享可靠的分布式事务，否则这两个独立系统之间存在双写窗口。

Transactional Outbox 不试图让两个系统“同时提交”。它把业务数据和待发送事件写进同一个 PostgreSQL 事务，再由后台 worker 至少一次地发布事件。接受重复投递，并让消费者幂等，是它的核心约束。

数据流

sequenceDiagram
    participant API as Order API
    participant DB as PostgreSQL
    participant W as Outbox Worker
    participant K as Kafka
    participant C as Consumer
    API->>DB: BEGIN
    API->>DB: UPDATE order + INSERT outbox_event
    API->>DB: COMMIT
    W->>DB: claim batch with lease
    W->>K: publish event_id + payload
    K-->>W: broker acknowledgement
    W->>DB: mark published
    K->>C: deliver (possibly more than once)
    C->>DB: insert inbox key + apply effect in one transaction

业务事务只负责本地数据库写入。Broker 不可用时，订单仍能提交，事件留在 Outbox 等待重试；反过来，业务事务回滚时，Outbox 记录也不会留下。

表结构与业务事务

事件应带稳定唯一 ID、事件类型、聚合键、版本化 payload 和创建时间。生产表通常还会有重试次数、下次可发送时间和租约字段：

CREATE TABLE outbox_event (
    event_id       uuid PRIMARY KEY,
    aggregate_type text NOT NULL,
    aggregate_id   text NOT NULL,
    event_type     text NOT NULL,
    payload        jsonb NOT NULL,
    created_at     timestamptz NOT NULL DEFAULT now(),
    available_at   timestamptz NOT NULL DEFAULT now(),
    attempts       integer NOT NULL DEFAULT 0,
    lease_token    uuid,
    lease_until    timestamptz,
    published_at   timestamptz
);

CREATE INDEX outbox_ready_idx
    ON outbox_event (available_at, created_at, event_id)
    WHERE published_at IS NULL;

服务层先保存订单，再插入事件；两者通过同一个 Spring 管理的事务提交：

@Transactional(rollbackFor = Exception.class)
public UUID confirmOrder(UUID orderId) throws JsonProcessingException {
    Order order = orders.require(orderId);
    order.confirm();
    orders.save(order);

    UUID eventId = UUID.randomUUID();
    outbox.insert(new OutboxEvent(
        eventId,
        "Order",
        order.id().toString(),
        "OrderConfirmed.v1",
        json.writeValueAsString(OrderConfirmed.from(order))
    ));
    return eventId;
}

此处假定 orders 和 outbox 使用同一数据源与本地事务管理器。Spring 的声明式事务依赖代理；同类内部自调用等情况可能绕过代理，需检查真实调用路径。

多 worker 领取事件

多个 worker 可用 FOR UPDATE SKIP LOCKED 避免同时领取相同行。以下片段使用 Spring NamedParameterJdbcTemplate 的具名参数形式，在短事务中选取一批事件并写入租约；事务提交后再调用 Kafka：

WITH picked AS (
    SELECT event_id
    FROM outbox_event
    WHERE published_at IS NULL
      AND available_at <= now()
      AND (lease_until IS NULL OR lease_until < now())
    ORDER BY available_at, created_at, event_id
    LIMIT :batch_size
    FOR UPDATE SKIP LOCKED
)
UPDATE outbox_event AS e
SET lease_token = :lease_token,
    lease_until = now() + interval '30 seconds',
    attempts = attempts + 1
FROM picked
WHERE e.event_id = picked.event_id
RETURNING e.*;

PostgreSQL 文档说明，SKIP LOCKED 会跳过当前锁住的行，因此适合多消费者访问队列表；它会给出不一致视图，不应当用作一般查询的一致性语义。

Worker 在事务外逐条发布；收到 broker 确认后，用 event_id 和本次 lease_token 条件更新 published_at。发布失败时，清空租约并设置指数退避后的 available_at。租约要长于正常发送时间，或通过续租机制延长；否则慢 worker 与新 worker 可能重复发送同一事件。

重复投递如何收敛

Outbox 无法消除这个窗口：Kafka 已确认消息，但 worker 在更新 published_at 前崩溃。租约过期后，另一 worker 会再次发送。因此端到端语义是 至少一次，不是恰好一次。

消费者可在同一个数据库事务中先写 Inbox 去重键，再执行本地业务变化：

INSERT INTO consumed_event (consumer_name, event_id)
VALUES (:consumer, :event_id)
ON CONFLICT DO NOTHING;

只有插入成功的事务才继续应用业务副作用。若副作用在另一个外部系统中，Inbox 仍需配合对方支持的幂等键或补偿流程。

实施检查与踩坑

不要在持有行锁时等待 Kafka。 先短事务领取并提交租约，网络调用放在事务外。
不要把消息发送和标记发布塞进一个长事务。 数据库无法回滚已经被 Broker 接受的消息。
按聚合维持顺序时要显式设计。 多 worker 的并行领取可能改变同一订单事件的顺序，可按 aggregate_id 分区或限制同一聚合并发。
给事件 payload 版本。 事件的生产者和消费者可能独立部署、独立演进。
监控最老未发布事件和重试数。 仅监控 worker 进程存活无法发现队列积压。
设计清理和保留策略。 归档已发布事件，并根据合规与排障需求决定保留时间。

数据流

表结构与业务事务

多 worker 领取事件

重复投递如何收敛

实施检查与踩坑

参考资料

相关文章