24.7 事务消息 (半消息 / 回查)

一句话

事务消息把 “发消息” 拆成两段：先投一条消费者看不见的半消息，本地事务成功后再把它转正；如果转正的确认丢了，Broker 就回查生产者的本地事务状态，替它做决定。

和本地消息表的分工

两者解决的是同一个洞——“本地事务 + 发消息” 不原子，但把可靠性责任放在了不同地方：

维度	本地消息表（24.6）	事务消息
可靠性责任	业务方：自己写 relay、重试、清理	MQ：Broker 存半消息、定时回查
消息存放在哪	业务库的一张表	MQ 的 half topic
投递延迟	轮询间隔（100ms–1s）	转正后立即投递，毫秒级
依赖	任意 MQ	必须支持事务消息的 MQ
业务侵入	建表 + relay + 归档	实现两个回调 + 本地事务表
主要运维风险	表膨胀、relay 积压	半消息堆积、回查失败

🔑 事务消息没有让问题消失，只是把 “消息表” 从你的库搬到了 MQ，把 “relay” 换成了 Broker 的回查线程。业务侧仍然要一张本地事务表——否则回查无法回答。

完整流程：两次投递 + 一次兜底回查

Producer                              Broker                        Consumer
   │  ① 发半消息（Prepared）            │
   ├──────────────────────────────────▶│ 写入 half topic，对消费组不可见
   │  ② 返回「存储成功」                 │
   │◀──────────────────────────────────┤
   │  ③ 执行本地事务（订单落库 + 状态表） │
   │  ④ Commit / Rollback               │
   ├──────────────────────────────────▶│ Commit → 写入真实 topic ──▶ 可见
   │                                    │ Rollback → 删除半消息
   │      （④ 丢失，或返回 UNKNOWN）      │
   │  ⑤ 回查 CheckLocalTransaction  ◀────┤ 定时扫描超时的半消息
   │  ⑥ 返回 Commit / Rollback / UNKNOWN │

步骤	动作	失败后果
①	发半消息	发送失败，本地事务根本不执行
③	执行本地事务	失败则回滚，并回 ④ ROLLBACK
④	二次确认	丢失 → 半消息留在 half topic，等回查
⑤	回查	生产者查本地事务表回答；查不出就 UNKNOWN
⑥	转正 / 丢弃	转正后消费可见；超过上限丢弃并告警

半消息为什么必须对消费者不可见

半消息 (half message / prepared message) 已经写进 Broker 的存储，但不在真实 topic 里，消费组读不到。原因有三：

1. 时序：消息先到、本地事务后提交。若此时可见，消费者可能先读到 “订单已创建”，回头查库却发现订单不存在。
2. 回滚：本地事务可能失败。可见的消息无法撤回，只能再发一条 “作废” 消息，把一次失败放大成两条消息。
3. 职责：半消息只是一张 “待定信封”，它的唯一用途是等生产者来盖章。盖章之前，它不构成业务事实。

💡 RocketMQ 4.x 把半消息存在 RMQ_SYS_TRANS_HALF_TOPIC，转正时把消息的 topic / queueId 恢复成真实值再写入真实 topic。5.x 服务端换成了事务 API，语义不变。

回查：把 “不确定” 收敛成 “确定”

Broker 不知道生产者那笔本地事务到底提交没有，于是定时来问：

参数（RocketMQ，实际以 broker.conf 为准）	默认值	含义
transactionTimeOut	6s	半消息超过这么久没收到二次确认，就开始回查
transactionCheckInterval	60s	两次回查之间的间隔
transactionCheckMax	15	最多回查多少次

按默认值算收敛时间：首次回查最早在第 6 秒，之后每 60 秒一次，第 15 次回查发生在 6 + 60 × 14 = 846s ≈ 14 分钟。超过上限仍无结论，半消息被丢弃 (部分版本转入 TRANS_CHECK_MAX_TIME_TOPIC 供人工排查)，并打日志告警。

⚠️ **回查不是重试业务。**回查只回答 “已经发生的那笔本地事务是什么结果”，绝不能借机再执行一次下单。否则一次回查就是一次重复下单——这是事务消息最致命的一类误用。

回查必须查库，不能查内存

生产者的回调长这样：

func (l *listener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	switch queryTxState(l.db, string(msg.Body)) { // 查本地事务表，不是查内存
	case "COMMITTED":
		return primitive.CommitMessageState
	case "ROLLED_BACK":
		return primitive.RollbackMessageState
	default:
		return primitive.UnkonwnState
	}
}

queryTxState 读的是数据库，不是进程内 map——因为回查可能发生在另一个实例上 (生产者重启、扩缩容、宕机迁移)。内存里的状态在崩溃那一刻就没了。

⚠️ 回查读主库。读到延迟的从库会返回 “查不到”，进而答 UNKNOWN；如果积攒到回查上限，一条已经提交的本地事务对应的消息会被丢弃，造成永久不一致。

本地事务表：回查的唯一依据

CREATE TABLE tx_log (
  order_id   VARCHAR(64) PRIMARY KEY,
  state      VARCHAR(16) NOT NULL,   -- PREPARING / COMMITTED / ROLLED_BACK
  updated_at DATETIME    NOT NULL DEFAULT NOW()
) ENGINE=InnoDB;

-- 业务提交：订单与状态在同一个本地事务里
BEGIN;
INSERT INTO orders (order_id, user_id, amount, status) VALUES (?, ?, ?, 'PAID');
UPDATE tx_log SET state = 'COMMITTED', updated_at = NOW() WHERE order_id = ?;
COMMIT;

PREPARING 这一档不是多余的：它让回查能区分 “还没开始” 和 “已经回滚”。

• 只写 “提交 / 回滚” 两态：本地事务失败时表里根本没有行，回查只能一直答 UNKNOWN，白等 14 分钟。
• 先写 PREPARING (独立小事务提交)，再执行本地事务改状态：回查看到 PREPARING 且已超过 10 分钟，就可以判定为 “生产者崩了”，安全地回 ROLLBACK。

三种返回值，三种命运

返回值	含义	Broker 动作	什么时候用
COMMIT	本地事务已提交	半消息转正，投递给消费组	查到 COMMITTED
ROLLBACK	本地事务已回滚	删除半消息，永不投递	查到 ROLLED_BACK，或业务明确失败
UNKNOWN	还判断不了	等下一轮回查，超限丢弃	本地事务在途、主从延迟、记录缺失

🔑 UNKNOWN 是合法返回值，但每次返回它都在消耗回查预算。15 次预算用完，Broker 会替你做决定——丢弃。所以 UNKNOWN 只能用于 “马上就会有结果” 的场景。

可运行的半消息 / 回查模拟

用内存结构模拟 Broker 与生产者，跑通四种结局 (Go 1.21+，go run 直接跑)：

package main

import (
	"fmt"
	"sync"
)

const (
	statePrepared   = "PREPARED"
	stateCommitted  = "COMMITTED"
	stateRolledBack = "ROLLED_BACK"
	stateDiscarded  = "DISCARDED"
)

// localTx 是生产者侧的本地事务表：回查时唯一能回答「到底提交没有」的依据。
type localTx struct {
	mu     sync.Mutex
	states map[string]string // orderID -> COMMITTED / ROLLED_BACK
}

func (l *localTx) set(id, state string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.states[id] = state
}

func (l *localTx) get(id string) string {
	l.mu.Lock()
	defer l.mu.Unlock()
	if s, ok := l.states[id]; ok {
		return s
	}
	return "UNKNOWN"
}

type halfMsg struct {
	id     string
	body   string
	state  string
	checks int
}

// broker 模拟 MQ：半消息先落盘，转正后才进真实 topic。
type broker struct {
	mu        sync.Mutex
	halves    []*halfMsg
	topic     []string // 消费者可见的消息
	maxChecks int
}

func (b *broker) find(id string) *halfMsg {
	for _, h := range b.halves {
		if h.id == id {
			return h
		}
	}
	return nil
}

func (b *broker) sendHalf(id, body string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.halves = append(b.halves, &halfMsg{id: id, body: body, state: statePrepared})
	fmt.Printf("① 半消息 %s 落盘，消费者不可见\n", id)
}

func (b *broker) commit(id string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	h := b.find(id)
	if h == nil || h.state != statePrepared {
		return
	}
	h.state = stateCommitted
	b.topic = append(b.topic, h.body)
	fmt.Printf("   ✓ %s 转正，消费者可见\n", id)
}

func (b *broker) rollback(id string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	h := b.find(id)
	if h == nil || h.state != statePrepared {
		return
	}
	h.state = stateRolledBack
	fmt.Printf("   ✗ %s 丢弃（本地事务回滚）\n", id)
}

// checkBack 模拟 Broker 的定时回查：问生产者「这笔本地事务到底成没成」。
func (b *broker) checkBack(id string, query func(string) string) {
	b.mu.Lock()
	defer b.mu.Unlock()

	h := b.find(id)
	if h == nil || h.state != statePrepared {
		return
	}
	h.checks++
	fmt.Printf("? 第 %d 次回查 %s：", h.checks, id)

	switch query(id) {
	case stateCommitted:
		h.state = stateCommitted
		b.topic = append(b.topic, h.body)
		fmt.Println("本地已提交 → 转正")
	case stateRolledBack:
		h.state = stateRolledBack
		fmt.Println("本地已回滚 → 丢弃")
	default:
		if h.checks >= b.maxChecks {
			h.state = stateDiscarded
			fmt.Println("仍 UNKNOWN，达到上限 → 丢弃并告警")
		} else {
			fmt.Println("仍 UNKNOWN → 等下一轮")
		}
	}
}

func main() {
	tx := &localTx{states: make(map[string]string)}
	b := &broker{maxChecks: 3}

	fmt.Println("场景 1：正常两阶段")
	b.sendHalf("m1", "order-1 已创建")
	tx.set("m1", stateCommitted)
	b.commit("m1")

	fmt.Println("\n场景 2：本地事务成功，但提交确认丢了")
	b.sendHalf("m2", "order-2 已创建")
	tx.set("m2", stateCommitted)
	calls := 0
	b.checkBack("m2", func(id string) string {
		calls++
		if calls == 1 {
			return "UNKNOWN" // 模拟本地事务还在执行
		}
		return tx.get(id)
	})
	b.checkBack("m2", tx.get)

	fmt.Println("\n场景 3：本地事务失败，回查发现已回滚")
	b.sendHalf("m3", "order-3 已创建")
	tx.set("m3", stateRolledBack)
	b.checkBack("m3", tx.get)

	fmt.Println("\n场景 4：回查始终 UNKNOWN，超过上限")
	b.sendHalf("m4", "order-4 已创建")
	for i := 0; i < 3; i++ {
		b.checkBack("m4", tx.get)
	}

	fmt.Println("\n消费者可见的消息：", b.topic)
}

输出：

场景 1：正常两阶段
① 半消息 m1 落盘，消费者不可见
   ✓ m1 转正，消费者可见

场景 2：本地事务成功，但提交确认丢了
① 半消息 m2 落盘，消费者不可见
? 第 1 次回查 m2：仍 UNKNOWN → 等下一轮
? 第 2 次回查 m2：本地已提交 → 转正

场景 3：本地事务失败，回查发现已回滚
① 半消息 m3 落盘，消费者不可见
? 第 1 次回查 m3：本地已回滚 → 丢弃

场景 4：回查始终 UNKNOWN，超过上限
① 半消息 m4 落盘，消费者不可见
? 第 1 次回查 m4：仍 UNKNOWN → 等下一轮
? 第 2 次回查 m4：仍 UNKNOWN → 等下一轮
? 第 3 次回查 m4：仍 UNKNOWN，达到上限 → 丢弃并告警

消费者可见的消息： [order-1 已创建 order-2 已创建]

场景 2 是回查的价值所在：确认丢了，消息没丢。场景 4 是它的边界：回查也不是万能的，答不出结论的消息最终会被丢弃。

真实客户端骨架 (RocketMQ)

下面的骨架需要真实 Broker，不能单机跑，但 API 是按上游实现写的 (rocketmq-client-go/v2：监听器在创建生产者时绑定，回查入参是 MessageExt)：

package main

import (
	"context"
	"database/sql"
	"log"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	_ "github.com/go-sql-driver/mysql"
)

type orderTxListener struct {
	db *sql.DB
}

// 本地事务：订单落库 + 状态表置 COMMITTED，同一个事务。
// 生产实现应先独立提交一行 PREPARING，再在这里改成 COMMITTED，
// 这样回查才能区分「还没开始」和「已经回滚」。
func (l *orderTxListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	orderID := string(msg.Body)
	tx, err := l.db.Begin()
	if err != nil {
		return primitive.RollbackMessageState // 事务都开不了，业务必然没做
	}
	defer tx.Rollback()

	if _, err := tx.Exec("INSERT INTO orders(order_id, status) VALUES(?, 'PAID')", orderID); err != nil {
		return primitive.RollbackMessageState
	}
	_, err = tx.Exec(`INSERT INTO tx_log(order_id, state) VALUES(?, 'COMMITTED')
	                  ON DUPLICATE KEY UPDATE state='COMMITTED'`, orderID)
	if err != nil {
		return primitive.RollbackMessageState
	}
	if err := tx.Commit(); err != nil {
		return primitive.UnkonwnState // 提交结果未知，交给回查裁决，绝不在这里重试业务
	}
	return primitive.CommitMessageState
}

// 回查：只读本地事务表，绝不重复执行业务。注意入参是 MessageExt。
func (l *orderTxListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	var state string
	err := l.db.QueryRow("SELECT state FROM tx_log WHERE order_id=?", string(msg.Body)).Scan(&state)
	if err != nil {
		return primitive.UnkonwnState
	}
	switch state {
	case "COMMITTED":
		return primitive.CommitMessageState
	case "ROLLED_BACK":
		return primitive.RollbackMessageState
	default:
		return primitive.UnkonwnState
	}
}

func main() {
	db, err := sql.Open("mysql", "dsn")
	if err != nil {
		log.Fatal(err)
	}

	// 监听器在创建生产者时绑定，没有单独的 Register 步骤。
	p, err := rocketmq.NewTransactionProducer(
		&orderTxListener{db: db},
		producer.WithNameServer([]string{"127.0.0.1:9876"}),
		producer.WithGroupName("order-tx-group"),
		producer.WithRetry(2),
	)
	if err != nil {
		log.Fatal(err)
	}
	if err := p.Start(); err != nil {
		log.Fatal(err)
	}
	defer p.Shutdown()

	msg := primitive.NewMessage("order-topic", []byte("order-123"))
	res, err := p.SendMessageInTransaction(context.Background(), msg)
	if err != nil {
		log.Printf("发送事务消息失败：%v", err)
		return
	}
	log.Printf("事务消息结果：%v", res.State)
}

💡 库里的常量名确实拼作 primitive.UnkonwnState (上游笔误，不是排版错误)；另外两个是 CommitMessageState、RollbackMessageState。

生产端的五条约束

1. **本地事务要短。**半消息在等，事务跑得越久，6 秒后的回查越可能撞上 “还没提交”，只能答 UNKNOWN。
2. **本地事务表与业务同库同事务。**分开写，就会出现 “业务成功、状态没记” → 回查答不出 → 消息被丢弃。
3. **回查回调只读、无副作用、可并发。**Broker 会用多线程回查同一批消息，回调里加锁写库会把回查拖慢。
4. **失败时优先返回 ROLLBACK，而不是 UNKNOWN。**业务已知失败就明确回滚，别浪费 15 次回查预算。
5. **UNKNOWN 只留给 “马上有结果” 的情况。**长期 UNKNOWN 等于把决定权交给 Broker 的丢弃策略。

消费端仍然要幂等

事务消息只保证生产侧的 “本地事务 ⇔ 消息投递” 原子性，不保证消费一次：

• 转正后投递失败 → MQ 重投；
• 消费成功、ack 丢失 → MQ 重投；
• 消费组再均衡 → 部分消息重放。

所以消费端必须按 msg_id 去重 (幂等表 / 业务唯一键 / 状态机 CAS)，规则与本地消息表完全一致 (见 24.6)。

破除误区：Kafka 事务 ≠ 事务消息

误区	现实
「Kafka 事务能保证数据库与消息的原子性」	不能。Kafka 事务只覆盖 Kafka 内部的多个 topic-partition 写入 + consumer offset 提交；数据库写入不在事务边界内
「用了事务消息，消费端就不用幂等」	事务消息只保证生产侧不丢、转正原子；消费侧仍是 at-least-once
「用了事务消息，不需要本地事务表」	回查必须查状态，没有表就只能答 UNKNOWN
「一条事务消息能覆盖多个服务的写入」	只覆盖单个 producer 的本地事务 + 一条消息；多服务多写要用 Saga / TCC

Kafka 事务的正确用途是流处理 exactly-once：transactional.id + 幂等 producer + isolation.level=read_committed，让 “读—处理—写—提交位点” 成为一个原子单元。它不解决 “MySQL 写成功、Kafka 没发出去”。

🔑 需要 DB 与 Kafka 的原子性，仍然只有两条路：本地消息表 (outbox)+ CDC，或者业务侧接受最终一致 + 对账。事务消息是 MQ 厂商提供的 “outbox 托管版”，不是通用协议。

反模式

反模式	后果
本地事务没有状态表，回查永远 UNKNOWN	回查超限后半消息被丢弃，业务成功但下游永远收不到
在回查里重新执行下单 / 扣款	每次回查都是一次重复业务，直接资损
本地事务里做远程调用、批量操作	超过 6 秒触发回查，回查又查到未提交，UNKNOWN 循环
回查读从库	主从延迟导致「查不到」，已提交的消息被误丢弃
消费端不幂等	重投导致重复发货、重复加积分
以为 ROLLBACK 能撤回已投递消息	转正后的消息收不回，只能补偿
用一条事务消息做多服务多写	只覆盖单 producer 单消息，跨服务仍需 Saga / TCC
不监控半消息堆积	磁盘被未决半消息占满，Broker 拒绝写入，全站消息不可用

自检

Q1 半消息已经写进 Broker 了，为什么还要专门设计成 “消费者不可见”？如果可见，会出现哪两个具体故障？

▸ 答案：半消息写盘时，本地事务还没执行完，消息代表的业务事实尚未成立。若可见：① 消费者读到 “订单已创建”，回查业务库却没有这笔订单，只能报错或写入脏数据；② 本地事务失败时消息已经投出，无法撤回，只能再发一条 “作废” 消息，把一次失败放大成两条消息，且中间存在错误的业务窗口。半消息本质是 “待生产者盖章的信封”，盖章前不构成事实。

Q2 回查解决的是哪个时间窗口的问题？生产者如果没有本地事务表，会发生什么？

▸ 答案：解决的是 “Broker 保存了半消息、但没收到生产者 Commit/Rollback 确认” 的窗口——确认包在网络上丢了，或生产者执行完本地事务就崩了。Broker 不知道结果，只能来问。没有本地事务表，回查只能答 UNKNOWN，15 次 (约 14 分钟) 后消息被丢弃：若本地事务其实已提交，下游永远收不到，形成永久不一致；若已回滚，则白等 14 分钟并产生告警噪音。

Q3 团队说 “我们用了 Kafka 事务，所以数据库和消息是原子的，消费端也不用幂等”。指出两处错误。

▸ 答案：① Kafka 事务的边界只在 Kafka 内部 (多个分区写入 + offset 提交)，不包含数据库写入。DB 提交成功、Kafka 事务失败 (或反之) 依然不一致。② 事务消息/事务生产者都不改变消费侧语义，消费仍是 at-least-once；重投、再均衡都会导致重复消费，必须幂等。要 DB 与 Kafka 原子，用 outbox + CDC，或接受最终一致 + 对账。