RocketMQ事務消息解析！

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

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

單體架構下的事務

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

以創建訂單為例，假設下單后需要做兩個操作：

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

在單體架構下只需使用@Transactional開啟事務，就可以保證數據的一致性。

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

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

分布式架構下的事務

以下單流程為例，在分布式架構下的處理流程如下：

?訂單服務生成訂單。
發送訂單生成的MQ消息，積分服務訂閱消息，有新的訂單生成之后消費消息，增加對應的積分記錄。

普通MQ消息存在的問題

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

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

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

RocketMQ事務消息基礎流程

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

基本流程

?第一階段：
發送 Message，Half Message，即半事務消息。
此類型的 Message 是不會被 Consumer 消費。
第二階段：如果半事務消息投遞成功，則會開始執行本地事務。
分為如下三種 Case：
本地事務執行成功：
會向 Broker 發送 commit 消息，被 commit 過后的 Message 才能被 Consumer 消費到。
本地事務執行失敗：
會向 Broker 發送 rollback 消息，Broker 則會將剛剛投遞的半事務消息刪除，從而保證上下游數據的一致性。
如果 Producer 實例或者網絡出現了問題，Producer 沒能及時地將本地事務執行的結果通知 Broker。
Broker 會通過掃描發現某條 Message 長時間處于半事務消息狀態。
Broker 會主動地向 Producer 詢問此 Message 對應的事務狀態。

值得注意的是：

?
RocketMQ 并不會無休止的的信息事務狀態回查，默認回查 15 次。
如果 15 次回查還是無法得知事務狀態，RocketMQ 默認回滾該消息。

RocketMQ事務消息使用限制

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

RocketMQ事務消息基本原理

采用2PC兩階段設計。

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

源碼解讀

發送事務消息調用的是TransactionMQProducer的sendMessageInTransaction方法：

主要有以下幾個步驟：

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

參考

《RocketMQ技術內幕》

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

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

責任編輯：姜華來源：月伴飛魚

RocketMQ 消息類型事務消息