RocketMQ事務(wù)消息解析！

作者：日常加油站 2024-10-29 08:34:27

事務(wù)消息是 RocketMQ 提供的一種消息類型，支持在分布式場景下保障消息生產(chǎn)和本地事務(wù)的最終一致性。RocketMQ?采用了2PC的思想來實現(xiàn)了提交事務(wù)消息，同時增加一個補(bǔ)償邏輯來處理二階段超時或者失敗的消息。

單體架構(gòu)下的事務(wù)

在單體系統(tǒng)的開發(fā)過程中，假如某個場景下需要對數(shù)據(jù)庫的多張表進(jìn)行操作，為了保證數(shù)據(jù)的一致性，一般會使用事務(wù)，將所有的操作全部提交或者在出錯的時候全部回滾。

以創(chuàng)建訂單為例，假設(shè)下單后需要做兩個操作：

?在訂單表生成訂單。
在積分表增加本次訂單增加的積分記錄。

在單體架構(gòu)下只需使用@Transactional開啟事務(wù)，就可以保證數(shù)據(jù)的一致性。

@Transactional
    public void order() {
        String orderId = UUID.randomUUID().toString();
        // 生成訂單
        orderService.createOrder(orderId);
        // 增加積分
        creditService.addCredits(orderId);
    }

?
但在分布式架構(gòu)下，訂單系統(tǒng)和積分系統(tǒng)可能是兩個獨立的服務(wù)，此時就不能使用上述的方法開啟事務(wù)了，因為它們不處于同一個事務(wù)中。
在出錯的情況下，無法進(jìn)行全部回滾，只能對當(dāng)前服務(wù)的事務(wù)進(jìn)行回滾。
所以就有可能出現(xiàn)訂單生成成功但是積分服務(wù)增加積分失敗的情況（也可能相反），此時數(shù)據(jù)處于不一致的狀態(tài)。

分布式架構(gòu)下的事務(wù)

以下單流程為例，在分布式架構(gòu)下的處理流程如下：

?訂單服務(wù)生成訂單。
發(fā)送訂單生成的MQ消息，積分服務(wù)訂閱消息，有新的訂單生成之后消費消息，增加對應(yīng)的積分記錄。

普通MQ消息存在的問題

?假如訂單創(chuàng)建成功，MQ消息發(fā)送成功，但是order方法在返回的前一刻，服務(wù)突然宕機(jī)。
由于開啟了事務(wù)，事務(wù)還未提交（方法結(jié)束后才會正常提交）。
所以訂單表并未生成記錄，但是MQ卻已經(jīng)發(fā)送成功并且被積分服務(wù)消費，此時就會存在訂單未創(chuàng)建但是積分記錄增加的情況。
假如先發(fā)送MQ消息再創(chuàng)建訂單，如果MQ消息發(fā)送成功，創(chuàng)建訂單失敗，那么同樣處于不一致的狀態(tài)。

@Transactional
    public void order() {
        String orderId = UUID.randomUUID().toString();
        // 創(chuàng)建訂單
        Order order = orderService.createOrder(orderDTO.getOrderId());
        // 發(fā)送訂單創(chuàng)建的MQ消息
        sendOrderMessge(order);
        return;
    }

可以使用RocketMQ事務(wù)消息解決上述問題。

RocketMQ事務(wù)消息基礎(chǔ)流程

?Apache RocketMQ在4.3.0版中已經(jīng)支持分布式事務(wù)消息。
事務(wù)消息是 RocketMQ 提供的一種消息類型，支持在分布式場景下保障消息生產(chǎn)和本地事務(wù)的最終一致性。
RocketMQ采用了2PC的思想來實現(xiàn)了提交事務(wù)消息，同時增加一個補(bǔ)償邏輯來處理二階段超時或者失敗的消息。

基本流程

?第一階段：
發(fā)送 Message，Half Message，即半事務(wù)消息。
此類型的 Message 是不會被 Consumer 消費。
第二階段：如果半事務(wù)消息投遞成功，則會開始執(zhí)行本地事務(wù)。
分為如下三種 Case：
本地事務(wù)執(zhí)行成功：
會向 Broker 發(fā)送 commit 消息，被 commit 過后的 Message 才能被 Consumer 消費到。
本地事務(wù)執(zhí)行失?。?/li>
會向 Broker 發(fā)送 rollback 消息，Broker 則會將剛剛投遞的半事務(wù)消息刪除，從而保證上下游數(shù)據(jù)的一致性。
如果 Producer 實例或者網(wǎng)絡(luò)出現(xiàn)了問題，Producer 沒能及時地將本地事務(wù)執(zhí)行的結(jié)果通知 Broker。
Broker 會通過掃描發(fā)現(xiàn)某條 Message 長時間處于半事務(wù)消息狀態(tài)。
Broker 會主動地向 Producer 詢問此 Message 對應(yīng)的事務(wù)狀態(tài)。

值得注意的是：

?
RocketMQ 并不會無休止的的信息事務(wù)狀態(tài)回查，默認(rèn)回查 15 次。
如果 15 次回查還是無法得知事務(wù)狀態(tài)，RocketMQ 默認(rèn)回滾該消息。

RocketMQ事務(wù)消息使用限制

?
事務(wù)消息不支持延時消息和批量消息。
事務(wù)性消息可能不止一次被檢查或消費，所以消費者端需要做好消費冪等。
事務(wù)消息的生產(chǎn)者 ID 不能與其他類型消息的生產(chǎn)者 ID 共享。
與其他類型的消息不同，事務(wù)消息允許反向查詢、MQ服務(wù)器能通過它們的生產(chǎn)者 ID 查詢到消費者。

RocketMQ事務(wù)消息基本原理

采用2PC兩階段設(shè)計。

?
將 Message 原本真實的 Topic 和 MessageQueue 進(jìn)行備份。
放入到PROPERTY_REAL_TOPIC、PROPERTY_REAL_QUEUE_ID中保存。
將消息投遞到一個內(nèi)部Topic中RMQ_SYS_TRANS_HALF_TOPIC，該隊列專門存儲事務(wù)消息。
所有的 Half Message 全部都寫入到 queueId 為 0 的 MessageQueue。
因為一個 Topic 下只有 1 個 MessageQueue：
這個 Topic 下的所有 Message 就是全局有序的，它們會按照先來后到的順序被消費。
如果本地事務(wù)執(zhí)行成功進(jìn)行Commit，則將RMQ_SYS_TRANS_HALF_TOPIC 隊列中的消息投遞到真實的Topic中，供后續(xù)流程執(zhí)行。
并刪除這條 Half Message ，但刪除也是假刪除，只是給 Message 打上一個刪除的 Tag。
如果本地事務(wù)執(zhí)行失敗進(jìn)行rollback，則直接刪除這條 Half Message ，但刪除也是假刪除。
如果本地事務(wù)遲遲沒有返回結(jié)果 (默認(rèn)時間是6s)，則會觸發(fā)事務(wù)回查機(jī)制
執(zhí)行回查之前需要校驗檢查次數(shù)是否到達(dá)了最大值（需要手動設(shè)置，沒有默認(rèn)值）。
或者是當(dāng)前 Half Message 存在是否超過了 Message 保存的上限，即 3天。
如果滿足上面條件中的一種Half Message 會被放進(jìn) TRANS_CHECK_MAX_TIME_TOPIC Topic 當(dāng)中。
一旦判定為需要執(zhí)行事務(wù)回查邏輯，那么當(dāng)前這條 Half Message 就算已經(jīng)被消費了。
在沒達(dá)到最大的校驗次數(shù)之前，都還需要將其投遞到事務(wù)隊列當(dāng)中，以便下次重試時再次執(zhí)行 Check 邏輯。
如果回查成功則刪除投遞的 Half Message。

源碼解讀

發(fā)送事務(wù)消息調(diào)用的是TransactionMQProducer的sendMessageInTransaction方法：

主要有以下幾個步驟：

?獲取事務(wù)監(jiān)聽器TransactionListener，如果獲取為空或者本地事務(wù)執(zhí)行器LocalTransactionExecuter為空將拋出異常。
因為需要通過TransactionListener或者LocalTransactionExecuter來執(zhí)行本地事務(wù)，所以不能為空。
在消息中設(shè)置prepared屬性，此時與普通消息（非事務(wù)消息）相比多了PROPERTY_TRANSACTION_PREPARED屬性。
調(diào)用send方法發(fā)送prepared消息也就是half消息，發(fā)送消息的流程與普通消息一致。
根據(jù)消息的發(fā)送結(jié)果判斷：
如果發(fā)送成功執(zhí)行本地事務(wù)，并返回本地事務(wù)執(zhí)行結(jié)果狀態(tài)，如果返回的執(zhí)行狀態(tài)結(jié)果為空，將本地事務(wù)狀態(tài)設(shè)置為UNKNOW。
發(fā)送成功之外的其他情況，包括FLUSH_DISK_TIMEOUT刷盤超時、FLUSH_SLAVE_TIMEOUT和SLAVE_NOT_AVAILABLE從節(jié)點不可用三種情況。
此時意味著half消息發(fā)送失敗，本地事務(wù)狀態(tài)置為ROLLBACK_MESSAGE回滾消息。
調(diào)用endTransaction方法結(jié)束事務(wù)。

參考

《RocketMQ技術(shù)內(nèi)幕》

https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md。

https://github.com/apache/rocketmq/blob/master/docs/cn/design.md。

責(zé)任編輯：姜華來源：月伴飛魚

RocketMQ 消息類型事務(wù)消息

自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡