<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

@Service
@RocketMQTransactionListener(txProducerGroup = "tx-order-group") // 與發(fā)送方組名一致
public class OrderTransactionListenerImpl implements RocketMQLocalTransactionListener {

    @Autowired
    private OrderService orderService;
    
    /**
     * 執(zhí)行本地事務(wù)
     * @param msg 收到的消息
     * @param arg 調(diào)用sendMessageInTransaction時傳入的額外參數(shù)
     * @return 事務(wù)狀態(tài)
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 從消息頭或arg中獲取業(yè)務(wù)ID，如訂單ID
        String orderId = (String) msg.getHeaders().get("orderId");
        try {
            // 執(zhí)行本地業(yè)務(wù)邏輯，例如：更新訂單狀態(tài)為“支付成功”
            boolean success = orderService.updateOrderStatus(orderId, OrderStatus.PAID);
            // 根據(jù)執(zhí)行結(jié)果返回提交或回滾
            return success ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
        } catch (Exception e) {
            // 記錄日志，返回UNKNOWN狀態(tài)，等待Broker回查
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    /**
     * 事務(wù)回查方法
     * @param msg 收到的消息
     * @return 事務(wù)狀態(tài)
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String orderId = (String) msg.getHeaders().get("orderId");
        // 根據(jù)orderId查詢數(shù)據(jù)庫，確認(rèn)本地事務(wù)的最終狀態(tài)
        OrderStatus status = orderService.queryOrderStatus(orderId);
        if (OrderStatus.PAID.equals(status)) {
            // 本地事務(wù)已成功，提交消息
            return RocketMQLocalTransactionState.COMMIT;
        } else if (OrderStatus.FAILED.equals(status)) {
            // 本地事務(wù)已失敗，回滾消息
            return RocketMQLocalTransactionState.ROLLBACK;
        } else {
            // 狀態(tài)仍不明確，繼續(xù)等待下次回查
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class OrderService {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public void payOrder(String orderId) {
        // 1. 構(gòu)建消息
        Message<String> message = MessageBuilder.withPayload("訂單支付成功，增加積分")
                .setHeader("orderId", orderId) // 設(shè)置業(yè)務(wù)ID，用于回查
                .build();
        
        // 2. 發(fā)送事務(wù)消息
        // 參數(shù)1: 事務(wù)組名（需與監(jiān)聽器內(nèi)txProducerGroup一致）
        // 參數(shù)2: 主題（Topic）
        // 參數(shù)3: 消息體
        // 參數(shù)4: 可選參數(shù)，會傳遞給executeLocalTransaction方法的arg參數(shù)
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("tx-order-group", "order-topic", message, orderId);
        
        System.out.println("發(fā)送結(jié)果：" + result.getSendStatus());
    }
}

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order-consumer-group")
public class OrderConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        // 1. 解析消息，獲取訂單ID
        // 2. 【關(guān)鍵】冪等性校驗(yàn)：查詢數(shù)據(jù)庫或Redis，判斷該訂單的積分是否已經(jīng)添加過
        // if (已處理) { return; }
        
        // 3. 執(zhí)行業(yè)務(wù)邏輯（例如，為用戶增加積分）
        // creditService.addCredit(...);
        
        // 4. 記錄處理狀態(tài)，標(biāo)記該消息已處理
    }
}