本地消息表 + RabbitMQ 最终一致性实战：可靠投递与消费幂等

分类：架构与高并发 · 作者：管理员 · 发布于 2026-10-21 · 阅读 0 · 标签：本地消息表 / RabbitMQ / 最终一致性 / 消息队列 / 分布式

一、场景与痛点

下单成功后要扣库存、发积分、同步搜索、通知运营。如果业务代码和"发消息"写在一起，会遇到两类经典故障：

先发消息、后提交事务：消息已经发出去，数据库却回滚了。下游照常扣了库存、发了积分，产生脏数据，且很难排查。
先提交事务、后发消息：MQ 临时抖动导致发送失败。业务成功了，下游却永远不知道，两边数据悄悄不一致。

很多同学用 try/catch 包住发送逻辑，失败就回滚业务。这并不能解决问题：MQ 的"发送成功"不等于"消费成功"，生产者和 Broker 之间天然存在"消息已到达但响应丢失"的窗口，回滚业务只会让系统更脆弱。

正确做法是把"要发出去的消息"变成一条必须落库的数据，与业务数据同生共死，再由一个独立投递者可靠推送。这就是本地消息表（Outbox）模式。

二、方案设计：本地消息表原理

核心思路：把消息记录和业务数据放进同一个数据库事务，先保证消息一定落库；之后由投递组件把消息可靠地推到 MQ；消费端用幂等兜住 MQ 的重复投递。整体可靠性不依赖任何一次 RPC 调用是否成功，而是依赖数据库事务 + 定时扫描 + 对账。

消息表结构（以 PostgreSQL 为例）：

CREATE TABLE biz_outbox (
    id             BIGSERIAL PRIMARY KEY,
    msg_id         VARCHAR(64)  NOT NULL UNIQUE,  -- 全局唯一消息 ID
    biz_type       VARCHAR(32)  NOT NULL,         -- order.created / point.granted ...
    biz_id         VARCHAR(64)  NOT NULL,         -- 业务主键，如订单号
    payload        JSONB        NOT NULL,         -- 完整消息体（业务快照）
    status         SMALLINT     NOT NULL DEFAULT 0, -- 0待发送 1已投递 2完成 9失败
    retry_count    INT          NOT NULL DEFAULT 0,
    next_retry_time TIMESTAMP   NOT NULL DEFAULT now(),
    created_at     TIMESTAMP    NOT NULL DEFAULT now(),
    sent_at        TIMESTAMP
);
CREATE INDEX idx_outbox_status_time ON biz_outbox (status, next_retry_time);

状态流转：0 待发送 → 1 已投递（发送成功）→ 2 完成（消费端确认）；连续失败超过阈值置为 9，进入人工补偿流程。

三、落地代码

1. 业务事务内写消息

关键点：投递动作放在事务提交后的 afterCommit 回调里，避免长事务期间消息被定时任务提前扫到、下游读到未提交数据。

@Transactional
public void createOrder(OrderCreateCmd cmd) {
    Order order = orderMapper.insert(cmd);
    OutboxMessage msg = OutboxMessage.of(
        UUID.randomUUID().toString().replace("-", ""),
        "order.created", order.getOrderNo(),
        Map.of("orderNo", order.getOrderNo(), "amount", order.getAmount()));
    outboxMapper.insert(msg); // 与订单同库同事务，要么都成功要么都回滚

    TransactionSynchronizationManager.registerSynchronization(
        new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                // 事务已提交，此刻投递最安全；即使失败也有兜底扫描重发
                mqTemplate.convertAndSend("order.exchange",
                    msg.getBizType(), msg);
                outboxMapper.markSent(msg.getMsgId());
            }
        });
}

2. 定时任务兜底投递

MQ 不可用、进程重启等情况下，afterCommit 里的发送可能失败。此时由定时任务扫描积压的待发送消息补投。

@Scheduled(fixedDelay = 30_000)
public void scanAndSend() {
    List<OutboxMessage> pending =
        outboxMapper.scanPending(100); // status=0 AND next_retry_time <= now
    for (OutboxMessage m : pending) {
        // 乐观锁 CAS：只有仍处于待发送状态才能抢到，
        // 防止与 afterCommit 或另一个扫描线程重复投递
        int updated = outboxMapper.compareAndSetStatus(m.getId(), 0, 1);
        if (updated == 0) continue;
        try {
            mqTemplate.convertAndSend("order.exchange", m.getBizType(), m);
            outboxMapper.markSent(m.getId());
        } catch (Exception e) {
            // 投递失败：指数退避，超过上限置为失败并告警
            int next = m.getRetryCount() + 1;
            outboxMapper.retryLater(m.getId(), next, calcNextRetry(next));
        }
    }
}

// UPDATE biz_outbox SET status=1, sent_at=now() WHERE id=? AND status=0
private static final String CAS_SQL =
    "UPDATE biz_outbox SET status=1, sent_at=now() " +
    "WHERE id=? AND status=0";

3. 消费端幂等（必做）

MQ 是 at-least-once 语义：生产者重试、消费者 ack 超时都会导致同一条消息被消费多次。幂等必须做在消费者这一侧，用消息 ID 作为去重键：

@RabbitListener(queues = "order.created.queue")
public void onOrderCreated(OutboxMessage msg) {
    String dedupKey = "dedup:order:" + msg.getMsgId();
    Boolean first = stringRedisTemplate.opsForValue()
        .setIfAbsent(dedupKey, "1", Duration.ofHours(24));
    if (Boolean.FALSE.equals(first)) {
        log.info("重复消息已忽略: {}", msg.getMsgId());
        return;
    }
    // 真正业务：扣库存、发积分、同步搜索……
    inventoryService.deduct(msg.getBizId(), msg.getPayload());
}

setIfAbsent 的过期时间要大于下游业务的最长处理时间，否则处理慢于 TTL 时重复消息会再次进来。

四、踩坑记录

事务未提交就被定时任务扫到：如果 insert 消息后不拦截，长事务里消息可能被提前投递，下游读到未提交数据。解法：投递放 afterCommit + 定时任务兜底，双保险。
重复投递不可避免，别指望 MQ 帮你保证：消费者必须幂等，且去重键要用 msg_id 而不是 biz_id——一个订单可能产生 order.created、point.granted 多条消息，用 biz_id 去重会互相误判。
状态流转的并发：定时任务和 afterCommit 可能同时操作同一条消息，必须用 UPDATE ... WHERE status=0 的 CAS 更新，直接 updateById 会导致重复发送。
重试风暴：MQ 不可用时，扫描任务每 30 秒拉 100 条、全部投递失败，瞬间打满重试队列。要加指数退避（1min/5min/30min/2h...）和单批上限，失败超过 5 次置为 status=9 转人工，而不是无限重试。
只有投递没有对账等于裸奔：要加一个对账任务，定期比对 MQ 已确认的消息与 biz_outbox 状态为 2 的记录，找出下游漏处理的单子补偿，否则"消息发了但消费端业务异常"永远发现不了。
消息表无限膨胀：status=2 的已完成消息要及时归档（如 7 天前迁到历史表），否则表越来越大，扫描越来越慢。

五、最佳实践与总结

消息体携带完整业务快照（订单号、金额、时间），下游消费不查主库也能处理，对账成本低。
与 Binlog CDC（Debezium + Outbox）方案对比：本地消息表无额外组件、落地简单，适合中小规模；CDC 不依赖应用层写消息表，但引入 Canal/Debezium 的运维成本更高。
监控三件套：待发送积压数、失败消息数、消费处理延迟，全部接告警。

本地消息表 + MQ 最终一致性不是最"先进"的方案，但它是把"可靠投递"和"业务事务"解耦的最朴素、最可控的工程解法。核心就三件事：同事务写消息、CAS 防重复投递、消费端幂等 + 定时对账兜底。把这三点做扎实，跨服务数据一致性就有了兜底，剩下的交给对账与告警。