SpringBoot整合RocketMq實現(xiàn)分布式事務

更新時間：2024年11月06日 10:33:39 作者：我是小趴菜

這篇文章主要為大家詳細介紹了SpringBoot整合RocketMq實現(xiàn)分布式事務的相關知識,文中的示例代碼講解詳細,有需要的小伙伴可以參考一下

MQ實現(xiàn)分布式事務原理

RocketMq提供了事務消息，要實現(xiàn)分布式事務主要還是利用它的事務消息

1：服務A首先會發(fā)送一條半事務的消息到MQ，此時服務接收方還是無法消費這條消息的
2：半事務消息發(fā)送成功之后，服務A開始執(zhí)行本地業(yè)務邏輯
3：服務A執(zhí)行完本地業(yè)務之后，提交事務，事務提交成功之后，MQ這條半事務消息就會變成原始可消費的消息
4：服務接收方這時候就可以消費到這條消息，繼續(xù)執(zhí)行后續(xù)業(yè)務了

那么在這整個過程中可能會出現(xiàn)的異常有哪些呢？

1：半事務消息發(fā)送失敗

如果半事務消息發(fā)送失敗，那么服務A就不會繼續(xù)執(zhí)行接下來的業(yè)務了，整個流程會直接退出

2：本地事務提交成功，發(fā)送COMMIT消息失敗

服務A事務提交之后，需要發(fā)送一條消息告訴MQ，這條半事務消息可以消費了，但是這時候，COMMIT消息發(fā)送失敗了，那么這條消息就還是處于半事務狀態(tài)，所以MQ會進行回查

回查服務A這個事務是否成功了，如果成功了，就會發(fā)送回查結果，如果本地事務成功了，那么回查就會發(fā)送COMMIT消息，這條消息重新設置為可消費狀態(tài)

實戰(zhàn)

服務-A

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

    //本地服務調用
    testDao.insert(test);

    //發(fā)送半事務消息
    //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;

    //執(zhí)行本地事務
    @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));
        
        //根據(jù)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));
        
        //還是根據(jù)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> {


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