RocketMQ事务消息最终一致性完整解决方案

sql
CREATE TABLE tx_transaction_log (
tx_id VARCHAR(64) NOT NULL COMMENT '全局事务ID(订单号)',
create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (tx_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='RocketMQ事务状态表(防空回滚占位)';

java
@Service
public class SeckillTxProducer {

@Autowired
private RocketMQTemplate rocketMQTemplate;

public void sendSeckillTxMessage(String orderNo, String json) {
Message<String> message = MessageBuilder.withPayload(json).build();
// 发送半消息
rocketMQTemplate.sendMessageInTransaction(
"seckill_tx_topic",
message,
orderNo
);
}
}

java
@Component
@RocketMQTransactionListener(producerGroup = "seckill_tx_group")
public class SeckillTxListener implements RocketMQLocalTransactionListener {

@Autowired
private OrderService orderService;
@Autowired
private TxLogMapper txLogMapper;

// ===================== 执行本地事务 =====================
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
String orderNo = (String) arg;
SeckillOrder order = JSON.parseObject(msg.getPayload().toString(), SeckillOrder.class);

try {
// 关键：插入事务占位记录
// 如果回查已经抢先插入，这里直接主键冲突
txLogMapper.insertTxLog(orderNo);

// 执行业务：创建订单 + 扣库存（本地事务）
orderService.createOrderAndDeductStock(order);

// 业务成功，提交消息
return RocketMQLocalTransactionState.COMMIT;
} catch (DuplicateKeyException e) {
// 主键冲突 = 回查已经占位，本次事务作废
return RocketMQLocalTransactionState.ROLLBACK;
} catch (Exception e) {
return RocketMQLocalTransactionState.ROLLBACK;
}
}

// ===================== 事务回查（核心防BUG逻辑） =====================
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
SeckillOrder order = JSON.parseObject(msg.getPayload().toString(), SeckillOrder.class);
String orderNo = order.getOrderNo();

// 1. 查询事务记录
Integer count = txLogMapper.countTxLog(orderNo);

if (count == null || count == 0) {
// 【最核心代码】无记录 → 主动占位！！
try {
txLogMapper.insertTxLog(orderNo);
} catch (DuplicateKeyException e) {
// 并发回查插入，忽略
}
// 占位成功，继续回查，不提交、不回滚
return RocketMQLocalTransactionState.UNKNOWN;
}

// 2. 有记录 = 本地事务已执行过
boolean orderExist = orderService.isOrderExist(orderNo);
if (orderExist) {
return RocketMQLocalTransactionState.COMMIT;
} else {
return RocketMQLocalTransactionState.ROLLBACK;
}
}
}

java
@Service
public class OrderService {

@Autowired
private OrderMapper orderMapper;
@Autowired
private StockMapper stockMapper;

@Transactional(rollbackFor = Exception.class)
public void createOrderAndDeductStock(SeckillOrder order) {
// 创建订单
orderMapper.insert(order);
// 扣减库存
int rows = stockMapper.deductStock(order.getGoodsId());
if (rows <= 0) {
throw new RuntimeException("库存不足");
}
}

public boolean isOrderExist(String orderNo) {
return orderMapper.selectByOrderNo(orderNo) != null;
}
}

java
@Mapper
public interface TxLogMapper {
@Insert("INSERT INTO tx_transaction_log(tx_id) VALUES(#{txId})")
void insertTxLog(@Param("txId") String txId);

@Select("SELECT count(1) FROM tx_transaction_log WHERE tx_id = #{txId}")
Integer countTxLog(@Param("txId") String txId);
}

java
@Component
@RocketMQMessageListener(topic = "seckill_tx_topic", consumerGroup = "seckill_consumer_group")
public class SeckillConsumer implements RocketMQListener<String> {

@Autowired
private OrderService orderService;

@Override
public void onMessage(String message) {
SeckillOrder order = JSON.parseObject(message, SeckillOrder.class);
// 幂等判断
if (!orderService.isOrderExist(order.getOrderNo())) {
return;
}
// 执行下游业务：通知、积分、物流等
}
}