@Transactional(rollbackFor = Exception.class)
@Override
public String mqInsert(Test test) {

    //本地服務(wù)調(diào)用
    testDao.insert(test);

    //發(fā)送半事務(wù)消息
    //Test是我們本地需要保存的一個(gè)對(duì)象
    Message<String> message = MessageBuilder.withPayload(JSONObject.toJSONString(test)).build();
    rocketMQTemplate.sendMessageInTransaction("test-topic", message, null);

    return "success";
}

@RocketMQTransactionListener
public class TransactionMqListener implements RocketMQLocalTransactionListener {

    @Resource
    private TestDao testDao;

    //執(zhí)行本地事務(wù)
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        //獲取半事務(wù)消息
        Test test = JSONObject.parseObject(new String((byte[]) message.getPayload()), Test.class);
        System.out.println("test | executeLocalTransaction | 消息是：" + JSONObject.toJSONString(test));
        
        //根據(jù)id查詢?cè)撚涗浭欠癖４娉晒α?
        Test testExist = testDao.queryById(test.getId());
        if(testExist == null) {
            //說(shuō)明本地事務(wù)提交失敗了，需要回滾
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        
        //本地事務(wù)提交成功
        return RocketMQLocalTransactionState.COMMIT;
    }

    //事務(wù)回查
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        Test test = JSONObject.parseObject(new String((byte[]) message.getPayload()), Test.class);
        System.out.println("test | checkLocalTransaction | 消息是：" + JSONObject.toJSONString(test));
        
        //還是根據(jù)id去查詢記錄
        Test testExist = testDao.queryById(test.getId());
        if(testExist == null) {
            //不存在，說(shuō)明本地事務(wù)提交失敗，回滾
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        
        //本地事務(wù)提交成功了
        return RocketMQLocalTransactionState.COMMIT;
    }
}

@Component
@RocketMQMessageListener(topic = "test-topic",consumerGroup = "cpy-consumer-group")
public class CpyListener implements RocketMQListener<String> {


    //消息監(jiān)聽(tīng)
    @Override
    public void onMessage(String s) {
        System.out.println("cpy服務(wù)收到消息：" + JSONObject.toJSONString(s));
    }
}