大宗商品销采运储系统・六层架构连载公共中间件层上一篇消息队列上篇我们学习了消息队列基础模型、流转链路以及 RocketMQ 基础接入代码实现业务事件异步投递。但在大宗商品销采运储业务里销售开单、库存扣减、财务记账这类核心业务会遇到一个经典难题数据库事务和消息发送如何保证原子性业务库事务提交成功消息发送失败下游业务收不到事件单据断层消息发送成功数据库事务回滚下游收到消息业务却没有生成单据造成数据不一致。本篇聚焦两大核心事务消息原理与落地、业务幂等设计解决销采运储核心单据的数据一致性问题。一句话定调事务消息解决「数据库和消息的原子性」幂等解决「消息重复投递带来脏数据」。一、什么是事务消息业务场景边界大宗商品业务场景举例销售订单创建成功需要发送消息触发仓储生成出库单。理想状态订单入库成功消息一定发出订单入库失败消息一定不发出。普通消息无法做到这时就需要 RocketMQ 事务消息。事务消息执行流程生产者发送半消息到 BrokerBroker 暂存消息此时下游消费者无法消费这条消息。Broker 收到半消息成功后生产者执行本地数据库事务新增销售订单、扣减预占库存。本地事务执行完成生产者向 Broker 提交确认Commit / Rollback。CommitBroker 将消息转为可消费状态下游消费者可以拉取处理。RollbackBroker 删除这条半消息下游永远收不到。如果网络中断、服务宕机Broker 长时间没有收到确认会主动发起事务回查。生产者收到回查请求查询本地事务状态再回复 Broker 提交或回滚。边界说明事务消息保证最终一致性不是强一致性。适合销采运储单据异步联动场景不适合要求实时强一致的业务。二、销采运储事务消息代码示例Spring Boot RocketMQpom.xml依赖和上篇保持一致下面是事务消息核心代码。1. 事务监听器RocketMQTransactionListener Component public class OrderTransactionListener implements RocketMQTransactionListener { Autowired private OrderService orderService; /** * 执行本地事务 */ Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { OrderDTO dto JSON.parseObject(msg.getBody(), OrderDTO.class); try { // 执行本地事务创建销售订单预扣库存 orderService.createOrderAndLockStock(dto); // 本地事务成功通知 Broker 提交消息 return LocalTransactionState.COMMIT_MESSAGE; } catch (Exception e) { // 本地事务失败回滚消息 return LocalTransactionState.ROLLBACK_MESSAGE; } } /** * 事务回查接口Broker 长时间未收到确认时调用 */ Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { OrderDTO dto JSON.parseObject(msg.getBody(), OrderDTO.class); // 查询数据库订单状态判断本地事务结果 boolean exist orderService.checkOrderExist(dto.getOrderNo()); if (exist) { return LocalTransactionState.COMMIT_MESSAGE; } return LocalTransactionState.ROLLBACK_MESSAGE; } }2. 发送事务消息Autowired private RocketMQTemplate rocketMQTemplate; public void createSaleOrder(OrderDTO dto) { String topic supply_order_stock_topic; MessageString message MessageBuilder.withPayload(JSON.toJSONString(dto)).build(); // 发送事务消息 rocketMQTemplate.sendMessageInTransaction(topic, message, dto); }3. 事务消息配置要点rocketmq:name-server: 10.0.0.20:9876;10.0.0.21:9876;10.0.0.22:9876producer:group: supply-producer-groupsend-message-timeout: 3000retry-times-when-send-failed: 2工程规范事务消息只用于核心单据场景普通通知类事件不要使用增加系统复杂度。事务回查必须实现不能省略是宕机场景的兜底机制。本地事务尽量轻量化避免长时间锁库。事务消息生产者组必须全局唯一避免与其他生产者混淆。三、业务幂等解决消息至少一次投递带来的重复消费RocketMQ 的投递保障是至少一次投递。网络抖动、消费成功后 ACK 丢失、服务重启重平衡都会造成同一条消息重复投递。放到销采运储业务重复消费会导致重复生成出库单、重复扣库存、重复生成财务应收单属于严重线上故障。3.1 三种幂等实现方案对比方案适用业务场景优点局限性数据库唯一索引订单、出库单、财务流水实现简单数据库层面强约束只适合新增类单据更新场景不适用业务状态机判断单据状态流转待处理/已处理/完成适配单据多次状态变更需要维护单据状态业务代码稍复杂幂等表记录跨服务、复杂事件场景通用性最强可记录消息 ID、业务单号需要额外建表存在存储开销销采运储项目选型单据新增场景优先唯一索引单据状态流转使用状态机跨系统复杂事件使用独立幂等记录表。3.2 幂等表建表 SQL 示例CREATE TABLE t_idempotent_record ( id bigint(20) NOT NULL AUTO_INCREMENT, business_no varchar(64) NOT NULL COMMENT 业务单号, msg_id varchar(64) NOT NULL COMMENT 消息ID, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_business_no (business_no), UNIQUE KEY uk_msg_id (msg_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT幂等记录表;3.3 幂等消费伪代码示例Override public void onMessage(OrderStockDTO message) { String orderNo message.getOrderNo(); // 1. 查询幂等记录 IdempotentRecord record idempotentMapper.selectByBusinessNo(orderNo); if (record ! null) { // 已经处理过直接返回不执行业务 log.info(单据{}已处理跳过本次消费, orderNo); return; } // 2. 执行业务逻辑生成出库单 stockService.createOutBoundOrder(message); // 3. 写入幂等记录 IdempotentRecord newRecord new IdempotentRecord(); newRecord.setBusinessNo(orderNo); newRecord.setMsgId(message.getMsgId()); newRecord.setCreateTime(new Date()); idempotentMapper.insert(newRecord); }重要提醒幂等判断 业务处理 幂等记录写入建议放在同一个数据库事务中防止中间宕机导致幂等记录未落库。四、本篇小结本篇我们介绍了 RocketMQ 事务消息解决数据库与消息发送的原子性同时落地业务幂等方案解决消息重复投递的脏数据风险。事务消息保证事件“有没有发出去”幂等保证事件“不会重复处理”两者搭配支撑大宗商品订单、库存、财务核心单据的异步流转。记住一句话事务消息保发送业务幂等保消费两板斧配合单据流转才不丢、不重。但在生产环境还会遇到消息堆积、重试风暴、死信消息、消息监控告警等运维难题这些内容放在下篇。下篇预告消息队列架构设计下篇消息堆积、重试风暴、死信队列治理与集群监控告警实战。