消息中间件升级，先演练事务消息的回查

消息中间件升级先演练事务消息的回查组件大版本升级应把事务消息的补偿和回查作为单独的验证项而不是只比较吞吐和 API 是否能编译。在把核心消息中间件从 RocketMQ 4.9 升级到 5.1 后的首个切流日报警日志响成一片分布式事务消息在 Broker 端积压了超过 30 万条且TRANSACTION_CHECK_GROUP的反查次数全部达到上限并进入死信队列DLQ。更严重的是部分订单服务发送了半消息Half Message后遭遇网络抖动按预期本应由 Broker 发起回查Check来决定 Commit 还是 Rollback。结果在 RocketMQ 5.x 的架构下回查请求被错误地路由到了没有注册事务监听器的代理 Proxy 节点上导致事务状态长时间挂起。最终订单系统扣了款库存系统却始终没有收到发货通知两端数据严重对不上账1. RocketMQ 4.x 与 5.x 事务消息机制的底层演进坑点为了解决高性能与云原生无状态扩展问题RocketMQ 5.x 引入了全新的Proxy gRPC 架构和POP 消费模式。然而正是这种物理架构的重构打乱了 4.x 时代沿用多年的事务消息补偿Transaction Check逻辑4.x 的反查机制Broker 发现半消息超过transactionTimeout没收到二次确认会根据记录的 Client Channel直接向 Producer 客户端发起 Netty 长连接 RPC 调起checkLocalTransaction。5.x 的 Proxy 路由遮蔽在 5.x 的架构中Producer 客户端通常连接的是无状态的 Proxy 节点而非直接连接 Broker。如果 4.x 版本的 Client SDK 与 5.x 版本的 Broker 混用Broker 的 Check 请求无法跨越 Proxy 正确找到原 Producer 实例。事务悬挂Transaction Hanging当网络抖动导致 Commit/Rollback 确认指令丢失而 Broker 端的 Check 又因为客户端路由问题不断失败消息就会永久挂起直至被废弃。升级分布式中间件时如果只评估了吞吐量TPS和 API 兼容性却忽略了这种基于异步回调的补偿机制风险极易失控。2. RocketMQ 5.x 事务消息回查与悬挂保护流程为了防止事务消息挂起与重复提交必须重新梳理 5.x 架构下的 Half Message 存储、回查与防悬挂流转图3. 生产级防悬挂与幂等 TransactionListener 代码在 RocketMQ 升级过渡期Producer 端的TransactionListener必须具备防悬挂Anti-Hanging和双向幂等控制能力。以下是适配 RocketMQ 5.x 的生产级 Java 实现package com.example.distributed.transaction.rocketmq; import org.apache.rocketmq.client.producer.LocalTransactionState; import org.apache.rocketmq.client.producer.TransactionListener; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageExt; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Transactional; Component public class ProductionOrderTransactionListener implements TransactionListener { private static final Logger log LoggerFactory.getLogger(ProductionOrderTransactionListener.class); private final JdbcTemplate jdbcTemplate; public ProductionOrderTransactionListener(JdbcTemplate jdbcTemplate) { this.jdbcTemplate jdbcTemplate; } /** * 执行本地业务逻辑如订单扣款 */ Override Transactional public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { String businessKey msg.getKeys(); // 使用唯一业务单号作为 Key log.info(Executing local transaction for orderKey{}, businessKey); try { // 1. 插入事务悬挂记录表 (若回查先于本地事务到达以此阻断悬挂) jdbcTemplate.update(INSERT INTO sys_tx_hang_guard(tx_key, status, create_time) VALUES (?, IN_PROGRESS, NOW()), businessKey); // 2. 执行核心业务 SQL boolean success doOrderPayment(businessKey, arg); if (success) { // 更新事务防悬挂表为 SUCCESS jdbcTemplate.update(UPDATE sys_tx_hang_guard SET status SUCCESS WHERE tx_key ?, businessKey); return LocalTransactionState.COMMIT_MESSAGE; } else { jdbcTemplate.update(UPDATE sys_tx_hang_guard SET status FAILED WHERE tx_key ?, businessKey); return LocalTransactionState.ROLLBACK_MESSAGE; } } catch (Exception ex) { log.error(Local transaction execution failed for orderKey{}, businessKey, ex); // 抛出异常或返回 UNKNOW迫使 Broker 随后发起 Check return LocalTransactionState.UNKNOW; } } /** * Broker 发起的补偿反查逻辑 */ Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { String businessKey msg.getKeys(); log.warn(Broker triggered transaction check for orderKey{}, msgId{}, businessKey, msg.getMsgId()); try { // 查询防悬挂表状态 String status jdbcTemplate.queryForObject( SELECT status FROM sys_tx_hang_guard WHERE tx_key ?, String.class, businessKey ); if (SUCCESS.equals(status)) { return LocalTransactionState.COMMIT_MESSAGE; } else if (FAILED.equals(status)) { return LocalTransactionState.ROLLBACK_MESSAGE; } else { // 如果记录为 IN_PROGRESS 且过了一段时间说明 executeLocalTransaction 卡死或失败回滚返回 ROLLBACK return LocalTransactionState.ROLLBACK_MESSAGE; } } catch (org.springframework.dao.EmptyResultDataAccessException e) { // 核心防悬挂点回查到达时本地事务居然没有任何记录 // 说明 executeLocalTransaction 根本还没开始执行网络延迟导致 Half Message 响应比本地事务快 // 此时必须插入 ROLLBACK 标记防止后续 executeLocalTransaction 再次成功提交引发数据不一致 log.error(Transaction hanging detected! No local record for orderKey{}. Forcing ROLLBACK., businessKey); jdbcTemplate.update(INSERT INTO sys_tx_hang_guard(tx_key, status, create_time) VALUES (?, HANG_ROLLBACK, NOW()), businessKey); return LocalTransactionState.ROLLBACK_MESSAGE; } } private boolean doOrderPayment(String businessKey, Object arg) { // 实际订单扣款逻辑实现... return true; } }4. 现场排障与终端命令行诊断升级上线后遇到分布式事务消息未决积压时可以通过 RocketMQmqadmin运维工具快速进行诊断。在 JumpServer 跳板机上检查集群状态与 Half Message 堆积情况# 1. 检查 Broker 与 Proxy 物理集群状态及版本号是否匹配 mqadmin clusterList -n ${ROCKETMQ_NAMESRV_ADDR} # 2. 专门查询 RocketMQ 内置的半消息系统 Topic 堆积量 mqadmin topicStatus -n ${ROCKETMQ_NAMESRV_ADDR} -t RMQ_SYS_TRANS_HALF_TOPIC # 输出结果示例 # #Cluster Name #Broker Name #QueueId #Min Offset #Max Offset #Last Update Time # DefaultCluster broker-a 0 10450 40450 2026-08-18 16:30:11 # (显示有 30,000 条半消息积压未获得 Commit/Rollback 确认!)针对特定的异常 MessageId提取其 Track 追踪路径确认回查请求死在哪个节点# 3. 追踪特定分布式事务消息的消费轨迹与 Check 记录 mqadmin queryMsgById -n ${ROCKETMQ_NAMESRV_ADDR} -i ${MESSAGE_ID} # 4. 在 Producer 节点实时 grep 检查是否收到了 Broker/Proxy 的回查请求 grep checkLocalTransaction /var/log/rocketmqlogs/transaction_check.log | tail -n 20通过这一系列诊断命令能够精准捕捉到是因为旧版 Client 遗留的 Channel 注册失效导致 Proxy 无法将 Check 请求发还给应用进而定位出根因。5. 分布式组件升级的风险评估防线为了防止升级再次引发数据不对账的惨剧我们制定了分布式组件升级的三大评估铁律客户端与服务端 SDK 版本对齐原则升级中间件服务端如 Broker 4.x 升 5.x时必须同步升级 Producer/Consumer 的客户端 SDK。严禁跨大版本跨 4.x 与 5.x混用 SDK消除隐性协议兼容隐患。数据库级事务防悬挂兜底表所有依赖分布式事务消息的业务必须在 DB 中建有tx_hang_guard状态表。不管回查与本地事务谁先到达一律以数据库物理锁状态为准斩断悬挂链条。升级前双跑与回滚灰度演练在新版本大流量切流前必须在压测环境注入 10% 的网络丢包率人工制造消息 Commit 超时强制触发 Check 回查流程。只有回查补偿成功率达到 100%才允许推到生产环境。