24.6 本地消息表

一句话

把 “发消息” 变成 “往同一个库里写一行待发记录”，与业务数据在同一个本地事务里提交；再由后台任务把这行记录搬到 MQ。用一次本地事务，替掉一个分布式事务。

它堵的是哪一个洞

“本地事务 + 通知下游” 天生不是原子的。两种顺序都错：

顺序	失败点	后果
先发消息，后写库	写库失败 / 回滚	下游已经收到并执行，撤不回来 → 脏数据（多发）
先写库，后发消息	发消息失败 / 进程崩溃	上游成功、下游永远不知道 → 数据不一致（少发）

🔑 本地消息表的关键洞察：**能塞进本地事务的动作，就不需要分布式事务。**消息落库和业务落库同生共死，剩下的只是 “怎么把库里的消息可靠地搬出去”。

流程：一次本地事务 + 一个搬运工

业务请求
  │
  ├─ BEGIN（本地事务）
  │    ├─ 扣款 / 下单（业务表）
  │    └─ INSERT outbox(status=NEW)      ← 消息与业务同生共死
  │  COMMIT
  │
  └─ 异步 relay：扫 outbox → 发 MQ → 成功则 status=SENT
                                └─ 失败：retries++、next_retry_at 退避
                                        │
                        消费方：幂等去重 → 执行业务 → ack

步骤	谁做	做什么	失败怎么办
①	业务服务	本地事务里写业务表 + outbox	整体回滚，什么都没发生
②	relay	SELECT ... WHERE status=0 AND next_retry_at<=NOW() LIMIT N	—
③	relay	投递到 MQ	重试计数 + 退避，超过上限转死信
④	relay	投递成功后 UPDATE status=1	更新失败 → 下轮重发（消费端幂等兜底）
⑤	消费方	幂等表去重 → 执行业务	失败则 nack / 重试
⑥	消费方	写幂等记录 + ack	ack 丢失 → MQ 重投，仍被幂等挡住

表设计：两列索引决定成败

CREATE TABLE outbox (
  id            BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  msg_id        VARCHAR(64)  NOT NULL,          -- 业务生成，全局唯一
  topic         VARCHAR(128) NOT NULL,
  payload       JSON         NOT NULL,          -- 完整事件，消费方不反查生产库
  status        TINYINT      NOT NULL DEFAULT 0,-- 0 NEW / 1 SENT / 2 DEAD
  retries       INT          NOT NULL DEFAULT 0,
  next_retry_at DATETIME     NOT NULL DEFAULT NOW(),
  created_at    DATETIME     NOT NULL DEFAULT NOW(),
  UNIQUE KEY uk_msg_id (msg_id),
  KEY idx_pending (status, next_retry_at)
) ENGINE=InnoDB;

-- 消费端幂等表：同一个消费组内，一条消息只处理一次
CREATE TABLE consumed_msg (
  consumer_group VARCHAR(64) NOT NULL,
  msg_id         VARCHAR(64) NOT NULL,
  consumed_at    DATETIME    NOT NULL,
  PRIMARY KEY (consumer_group, msg_id)
);

四个设计点：

1. msg_id 业务侧生成，如 orderID:OrderPaid。它同时是发送侧去重键和消费侧幂等键——两边用同一个 ID，排查时能对上。
2. (status, next_retry_at) 联合索引：relay 的查询必须走索引且带 LIMIT，否则消息一多就是全表扫，直接打爆业务库。
3. payload 存完整事件：消费方只依赖消息体，不反查生产库。反查会引入新耦合，还会读到 “更新后的状态” 而不是 “事件发生时的状态”。
4. status 只增不改语义：NEW → SENT，重试超限 NEW → DEAD。不要物理删除 NEW 消息，删了就是丢消息。

💡 msg_id 的唯一键还顺带解决了 “业务重复提交”：同一笔订单重复支付，第二次插入直接撞唯一键，消息不会重复产生。

搬运工：轮询还是 CDC

维度	轮询 relay	CDC（binlog 订阅）
延迟	轮询间隔，100ms–1s	通常 < 100ms
对业务库压力	周期性查询，依赖索引	读 binlog，几乎不增加查询压力
额外组件	无	Debezium / Canal / Maxwell + Kafka
顺序性	单实例天然有序；多实例要按主键分片	按 binlog 顺序，天然有序
运维复杂度	低	中：权限、位点、schema 变更都要管
适合	中小流量、想少依赖	大流量、低延迟、已有 CDC 平台

⚠️ CDC 方案有个容易忽略的前提：binlog 只在事务提交后才可见。这正好是我们要的语义——业务回滚了，binlog 里也没有这条 outbox 记录。

投递语义：至少一次 + 幂等 = 效果一次

本地消息表不可能端到端 exactly-once，因为它有两个天然重复点：

• relay 投递成功、但 UPDATE status=1 失败 → 重启后重发；
• 消费方处理成功、但 ack 前崩溃 → MQ 重投。

所以正确的承诺是：at-least-once 投递 + 消费端幂等 = 效果上只处理一次。

消费端幂等的四种实现：

做法	说明	风险
幂等表（唯一键）	INSERT INTO consumed_msg ...，撞键就跳过	多一次写，需定期清理
业务唯一键	订单号、流水号本身唯一	依赖业务有天然唯一键
状态机 CAS	UPDATE ... WHERE status='待处理'	只适用于有状态流转的业务
Redis SETNX	快，但可能丢（过期、主从切换）	不能单独用于资金类

可运行的最小实现

下面用内存表模拟数据库，跑通 “投递失败重试 → 最终成功 → 重复消费被去重” (Go 1.21+，go run 直接跑)：

package main

import (
	"errors"
	"fmt"
	"sync"
)

const (
	statusNew  = 0 // 待发送
	statusSent = 1 // 已发送
	statusDead = 2 // 死信
)

type message struct {
	ID      string
	Topic   string
	Payload string
	Status  int
	Retries int
}

type store struct {
	mu       sync.Mutex
	accounts map[string]int64
	outbox   []*message
	applied  map[string]bool // 消费端幂等表：只认 msg_id
	notes    []string
}

func newStore() *store {
	return &store{
		accounts: map[string]int64{"alice": 1000, "bob": 0},
		applied:  make(map[string]bool),
	}
}

// transfer：扣款、入账、写消息表，三件事在同一个本地事务里完成。
func (s *store) transfer(from, to string, amount int64, msgID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.accounts[from] < amount {
		return errors.New("余额不足")
	}
	s.accounts[from] -= amount
	s.accounts[to] += amount
	s.outbox = append(s.outbox, &message{
		ID:      msgID,
		Topic:   "account.transferred",
		Payload: fmt.Sprintf("%s→%s %d 元", from, to, amount),
		Status:  statusNew,
	})
	return nil
}

// relay：扫描待发送消息并投递。真实实现要 LIMIT + (status, next_retry_at) 索引。
func (s *store) relay(pub func(*message) error, batch int) (sent int) {
	s.mu.Lock()
	defer s.mu.Unlock()

	for _, m := range s.outbox {
		if sent >= batch {
			break
		}
		if m.Status != statusNew {
			continue
		}
		if err := pub(m); err != nil {
			m.Retries++
			if m.Retries >= 3 {
				m.Status = statusDead
				fmt.Printf("☠ %s 重试 3 次仍失败，转死信并告警\n", m.ID)
			} else {
				fmt.Printf("✗ %s 投递失败（第 %d 次）：%v\n", m.ID, m.Retries, err)
			}
			continue
		}
		m.Status = statusSent
		sent++
		fmt.Printf("✓ %s 投递成功\n", m.ID)
	}
	return sent
}

// consume：消费端按 msg_id 去重后再执行业务。
func (s *store) consume(m *message) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.applied[m.ID] {
		fmt.Printf("↺ 消费 %s：重复消息，丢弃\n", m.ID)
		return
	}
	s.applied[m.ID] = true
	s.notes = append(s.notes, m.Payload)
	fmt.Printf("→ 消费 %s：%s（发积分 / 发通知）\n", m.ID, m.Payload)
}

func (s *store) show() {
	s.mu.Lock()
	defer s.mu.Unlock()
	fmt.Printf("账户：alice=%d bob=%d\n", s.accounts["alice"], s.accounts["bob"])
	for _, m := range s.outbox {
		fmt.Printf("消息 %s status=%d retries=%d\n", m.ID, m.Status, m.Retries)
	}
	fmt.Printf("下游已处理：%v\n", s.notes)
}

func main() {
	st := newStore()
	if err := st.transfer("alice", "bob", 100, "m-1"); err != nil {
		panic(err)
	}
	fmt.Println("本地事务提交完成（业务 + 消息一起落库）")
	st.show()

	fmt.Println("\n— relay 开始扫描，前两次 MQ 超时")
	queue := make([]*message, 0, 1)
	failLeft := 2
	pub := func(m *message) error {
		if failLeft > 0 {
			failLeft--
			return errors.New("MQ 超时")
		}
		cp := *m
		queue = append(queue, &cp)
		return nil
	}
	for i := 1; i <= 4; i++ {
		fmt.Printf("— 第 %d 轮 relay\n", i)
		if n := st.relay(pub, 10); n == 0 {
			fmt.Println("   （无新消息）")
		}
	}

	fmt.Println("\n— 消费：MQ 至少一次投递，同一条消息到达两次")
	for _, m := range queue {
		st.consume(m)
		st.consume(m)
	}

	fmt.Println()
	st.show()
}

输出：

本地事务提交完成（业务 + 消息一起落库）
账户：alice=900 bob=100
消息 m-1 status=0 retries=0
下游已处理：[]

— relay 开始扫描，前两次 MQ 超时
— 第 1 轮 relay
✗ m-1 投递失败（第 1 次）：MQ 超时
   （无新消息）
— 第 2 轮 relay
✗ m-1 投递失败（第 2 次）：MQ 超时
   （无新消息）
— 第 3 轮 relay
✓ m-1 投递成功
— 第 4 轮 relay
   （无新消息）

— 消费：MQ 至少一次投递，同一条消息到达两次
→ 消费 m-1：alice→bob 100 元（发积分 / 发通知）
↺ 消费 m-1：重复消息，丢弃

账户：alice=900 bob=100
消息 m-1 status=1 retries=2
下游已处理：[alice→bob 100 元]

三个观察点：钱只转了 100 (不是 200)；同一条消息到达两次，下游只处理一次；status 从 0 变成 1，retries=2 记录了两次失败。

容量：这张表会长得比你想象的快

• 写入量 = 业务写 TPS：每笔业务一条消息。1000 TPS 跑一天 ≈ 8640 万行。
• 空间：单行按 500 字节估算，一天约 43GB；加索引更大。
• relay 吞吐：批量 500 条、间隔 100ms ≈ 5000 条/秒的上限；要按峰值流量留 3–5 倍余量。

所以这张表必须当成流水表来治理：

措施	做法
只留必要字段	payload 里别塞整页 HTML / 大对象，存 ID + 关键字段
及时清理	SENT 超过 7 天直接删；DEAD 保留到人工处理完
分区	按 created_at 月分区，删分区比 DELETE 快几个数量级
监控积压	COUNT(*) WHERE status=0 + 最老 NEW 消息的年龄，两个指标都要告警
relay 多实例	按 id % N 分片，或 SELECT ... FOR UPDATE SKIP LOCKED

反模式

反模式	后果
outbox 与业务表分库	失去本地事务，业务成功、消息丢失（或反之）
先发消息、后写库	下游已执行，上游回滚 → 脏数据无法撤回
relay 查询不带 LIMIT / 不走索引	全表扫，业务库 CPU 打满，拖垮在线流量
发送成功却不更新状态	每轮重发，消费端被重复消息淹没
消费端不做幂等	重复扣款 / 重复发货，直接资损
无限重试、不设死信	一条毒消息卡住整个 relay，后面全部积压
outbox 永不清理	单表过亿后查询、备份、DDL 全部变慢
用本地消息表做「实时通知」，轮询 5s	用户等 5 秒才收到结果，体验不可接受

选型：三种 “最终一致” 怎么挑

方案	一致性强度	依赖	业务侵入	延迟	适合
本地消息表	不丢消息（最终一致）	任意 MQ	业务库多一张表 + relay	轮询间隔	已有业务库、想自己掌控可靠性
事务消息（24.7）	不丢消息（最终一致）	支持事务消息的 MQ	生产者实现回查	低，转正即投	已用 RocketMQ，不想改库
最大努力通知（24.8）	弱，可能丢	无	低	取决于重试	通知类，可容忍失败 + 对账

🔑 本地消息表是三个里最不依赖中间件的方案：不要求 MQ 支持事务、不要求 binlog 权限，一个定时任务就能跑。代价是业务库多一张持续增长的表。

自检

Q1 为什么 outbox 必须和业务表同库同事务？如果放到另一个库，会出现什么具体故障？

▸ 答案：同库同事务保证 “业务成功 ⇔ 消息落库” 是原子的。分库后变成两个独立事务：业务提交了、消息库写入失败，下游永远收不到 (少发)；反过来消息写入成功、业务回滚，下游会执行一笔不存在的业务 (多发)。这两种都需要额外的一致性机制去修，等于又绕回分布式事务。

Q2 relay 投递成功，但 UPDATE status=1 时数据库挂了。系统恢复后会发生什么？靠什么兜底？

▸ 答案：这一行仍是 NEW，下一轮 relay 会重复投递同一条消息。这不是 bug，是 at-least-once 的必然结果；兜底靠消费端按 msg_id 幂等 (幂等表 / 业务唯一键)。要减少重复窗口，可以把 “投递” 和 “改状态” 做得更近 (如同步更新、或用事务消息让 Broker 承担状态)，但无法消除。

Q3 下游已经做了幂等，为什么还必须监控 outbox 的积压量？

▸ 答案：幂等只解决 “重复”，不解决 “延迟” 和 “丢失”。积压意味着 relay 挂了、MQ 不可用或消费端卡住，此时业务已经提交、下游却迟迟收不到，用户看到的是 “钱扣了、权益没到”。另外 NEW 消息如果超过重试上限转成 DEAD，就再也不会被投递——只能靠积压监控 + 最老消息年龄告警发现。