/**
 * 發(fā)送事務(wù)消息
 *
 * @param topic   topic
 * @param message 消息對象
 */
public void sendMessageInTransaction(String topic, Object message) {
    String transactionId = UUID.randomUUID().toString();
    TransactionSendResult result = this.rocketMQTemplate.sendMessageInTransaction(topic, MessageBuilder.withPayload(message)
            .setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
            .build(), message);
}

@RocketMQTransactionListener
public class OrderTransactionListener implements RocketMQLocalTransactionListener {

    /**
     * 執(zhí)行本地事務(wù)，即處理核心鏈路
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
          	// 事務(wù)transactionId
            String transactionId = msg.getHeaders().get("rocketmq_TRANSACTION_ID").toString();
            // 本地事務(wù)，執(zhí)行核心鏈路業(yè)務(wù)
            String payload = new String((byte[]) msg.getPayload());
            OrderTranscationMesageDTO data = JSONObject.parseObject(payload, OrderTranscationMesageDTO.class);
            orderService.executeCoreBusiness(data.getPayMoney(),data.getOrderDO(), data.getTransactionNo(),data.getPayType(),transactionId);
        } catch (Exception e) {
            log.error("本地事務(wù)執(zhí)行異常:{}事務(wù)消息回滾", e.getMessage());
            return RocketMQLocalTransactionState.ROLLBACK;
        }
      	log.info("提交事務(wù)消息");
        return RocketMQLocalTransactionState.COMMIT;
    }

    /**
     * 校驗本地事務(wù)（broker未收到提交或回滾事務(wù)消息時主動回查）
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 事務(wù)transactionId
        String transactionId = msg.getHeaders().get("rocketmq_TRANSACTION_ID").toString();
        // 數(shù)據(jù)庫能查到，說明本地事務(wù)執(zhí)行失敗，需要回滾
        if (Objects.isNull(transcationLogDao.getById(transactionId))){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        return RocketMQLocalTransactionState.COMMIT;
    }

}

/**
  * 核心業(yè)務(wù)，并記錄RocketMQ事務(wù)ID
*/
@GlobalTransactional  // seata全局事務(wù)
public void executeCoreBusiness(BigDecimal payMoney, AppDerivativeGoodsOrderDO orderDO, String transactionNo, String payType,String transactionId) {
		// 核心業(yè)務(wù)偽代碼
    orderService.execute();
    storeFeign.execute();
    scoreFeign.execute();
    
    // 數(shù)據(jù)庫記錄rocket事務(wù)消息ID 用于異常情況下的回查
    if (Objects.nonNull(transactionId)){
        //寫入事務(wù)日志
        TransactionLogDO log = new TransactionLogDO();
        log.setId(transactionId);
        log.setBusiness("order");
        log.setForeignKey(String.valueOf(orderDO.getId()));
        transcationLogDao.save(log);
    }
}

CREATE TABLE `transaction_log` (
  `id` varchar(32) COLLATE utf8mb4_bin NOT NULL DEFAULT '' COMMENT '事務(wù)ID',
  `business` varchar(32) COLLATE utf8mb4_bin NOT NULL COMMENT '業(yè)務(wù)標(biāo)識',
  `foreign_key` varchar(32) COLLATE utf8mb4_bin NOT NULL COMMENT '對應(yīng)業(yè)務(wù)表中的主鍵',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

@RocketMQMessageListener(topic = "order_topic",consumerGroup = "order_group")
public class TestListener implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt messageExt) {
        if(messageExt.getReconsumeTimes() >= 3){
            log.error("消息重試已達(dá)最大次數(shù)，將通知業(yè)務(wù)人員排查問題。{}",messageExt.getMsgId());
            //消息處理，第3次處理失敗后，發(fā)送郵件通知人工介入
            sendMail(messageExt.getMsgId());
        }
        // 倉儲物流相關(guān)業(yè)務(wù) 
        wmsService.execute(messageExt.getBody());
    }
}