首页 > 软件编程 > java > SpringBoot RocketMq分布式事务

SpringBoot整合RocketMq实现分布式事务

2024-11-06 10:33:39 作者：我是小趴菜

这篇文章主要为大家详细介绍了SpringBoot整合RocketMq实现分布式事务的相关知识,文中的示例代码讲解详细,有需要的小伙伴可以参考一下

大家好，今天我们继续分布式事务的学习，之前我们已经实战了

springboot整合Atomikos

springboot整合Himly

来实现分布式事务，今天我们继续学习springboot整合RocketMq来实现分布式事务

MQ实现分布式事务原理

RocketMq提供了事务消息，要实现分布式事务主要还是利用它的事务消息

1：服务A首先会发送一条半事务的消息到MQ，此时服务接收方还是无法消费这条消息的
2：半事务消息发送成功之后，服务A开始执行本地业务逻辑
3：服务A执行完本地业务之后，提交事务，事务提交成功之后，MQ这条半事务消息就会变成原始可消费的消息
4：服务接收方这时候就可以消费到这条消息，继续执行后续业务了

那么在这整个过程中可能会出现的异常有哪些呢？

1：半事务消息发送失败

如果半事务消息发送失败，那么服务A就不会继续执行接下来的业务了，整个流程会直接退出

2：本地事务提交成功，发送COMMIT消息失败

服务A事务提交之后，需要发送一条消息告诉MQ，这条半事务消息可以消费了，但是这时候，COMMIT消息发送失败了，那么这条消息就还是处于半事务状态，所以MQ会进行回查

回查服务A这个事务是否成功了，如果成功了，就会发送回查结果，如果本地事务成功了，那么回查就会发送COMMIT消息，这条消息重新设置为可消费状态

实战

服务-A

@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;
    }
}

服务B

@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));
    }
}