SpringBoot整合RocketMq实现分布式事务

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

    //本地服务调用
    testDao.insert(test);

    //发送半事务消息
    //Test是我们本地需要保存的一个对象
    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;

    //执行本地事务
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        //获取半事务消息
        Test test = JSONObject.parseObject(new String((byte[]) message.getPayload()), Test.class);
        System.out.println("test | executeLocalTransaction | 消息是：" + JSONObject.toJSONString(test));
        
        //根据id查询该记录是否保存成功了
        Test testExist = testDao.queryById(test.getId());
        if(testExist == null) {
            //说明本地事务提交失败了，需要回滚
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        
        //本地事务提交成功
        return RocketMQLocalTransactionState.COMMIT;
    }

    //事务回查
    @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));
        
        //还是根据id去查询记录
        Test testExist = testDao.queryById(test.getId());
        if(testExist == null) {
            //不存在，说明本地事务提交失败，回滚
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        
        //本地事务提交成功了
        return RocketMQLocalTransactionState.COMMIT;
    }
}

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


    //消息监听
    @Override
    public void onMessage(String s) {
        System.out.println("cpy服务收到消息：" + JSONObject.toJSONString(s));
    }
}