2. 程式人生 > >RocketMq在閹割訊息回查checkTransactionState後實現分散式事務

RocketMq在閹割訊息回查checkTransactionState後實現分散式事務

阿新 • • 發佈：2019-01-02

利用rocketMQ解決分散式事務

在rocketMQ中生產者有三種角色 NormalProducer（普通）、OrderProducer（順序）、TransactionProducer（事務）
根據名字大概可以看出各個代表著什麼作用，我們這裡用 TransactionProducer（事務）來解決問題。

先舉個列子來說明下我們解決方案的設計方式吧：最經典的莫過於銀行轉賬了，網上到處都有，時序圖如下
這裡寫圖片描述

下面貼一下測試程式碼：

(1) 執行業務邏輯的部分

/**
 * @Date: Created in  2018/2/12 15:55
    執行本地事務
 */
public class
 
 TransactionExecuterimpl implements LocalTransactionExecuter{
    @Override
    public LocalTransactionState executeLocalTransactionBranch(final Message message, final Object o) {
        try{
        //DB操作 應該帶上事務 service -> dao
        //如果資料操作失敗  需要回滾    同事返回RocketMQ一個失敗訊息  意味著 消費者無法消費到這條失敗的訊息
        //如果成功 就要返回一個rocketMQ成功的訊息，意味著消費者將讀取到這條訊息
 

        //o就是attachment

            //測試程式碼
           if(new Random().nextInt(3) == 2){
                int a = 1 / 0;
           }
            System.out.println(new Date()+"===> 本地事務執行成功，傳送確認訊息");
        }catch (Exception e){
            System.out.println(new Date()+"===> 本地事務執行失敗！！！");
            return
 
 LocalTransactionState.ROLLBACK_MESSAGE;

        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

(2) 處理事務回查的程式碼部分

/**
 * @Date: Created in  2018/2/12 15:48
 * 未決事務，伺服器端回查客戶端
 */
public class TransactionCheckListenerImpl implements TransactionCheckListener {
    @Override
    public LocalTransactionState checkLocalTransactionState(MessageExt messageExt) {

        System.out.println("伺服器端回查事務訊息： "+messageExt.toString());
        //由於RocketMQ遲遲沒有收到訊息的確認訊息，因此主動詢問這條prepare訊息，是否正常？
        //可以查詢資料庫看這條資料是否已經處理

        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

(3) 啟動生產者

/**
 * @Date: Created in  2018/2/12 15:24
 * 測試本地事務
 */
public class TestTransactionProducer {
    public static void main(String[] args){
        //事務回查監聽器
        TransactionCheckListenerImpl checkListener = new TransactionCheckListenerImpl();
        //事務訊息生產者
        TransactionMQProducer producer = new TransactionMQProducer("transactionProducerGroup");
        //MQ伺服器地址
        producer.setNamesrvAddr("192.168.56.105:9876;192.168.106:9876");
        //註冊事務回查監聽
        producer.setTransactionCheckListener(checkListener);
        //本地事務執行器
        TransactionExecuterimpl executerimpl = null;
        try {
            //啟動生產者
            producer.start();
            executerimpl = new TransactionExecuterimpl();
            Message msg1 = new Message("TransactionTopic", "tag", "KEY1", "hello RocketMQ 1".getBytes());
            Message msg2 = new Message("TransactionTopic", "tag", "KEY2", "hello RocketMQ 2".getBytes());

            SendResult sendResult = producer.sendMessageInTransaction(msg1, executerimpl, null);
            System.out.println(new Date() + "msg1"+sendResult);

            sendResult = producer.sendMessageInTransaction(msg1, executerimpl, null);
            System.out.println(new Date() + "msg2"+sendResult);

        } catch (MQClientException e) {
            e.printStackTrace();
        }
        producer.shutdown();
    }
}

(4) 消費之消費訊息

/**
 * @Date: Created in  2018/2/11 15:37
 */
public class TestConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
        consumer.setNamesrvAddr("192.168.56.105:9876;192.168.56.106:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //消費普通訊息
       // consumer.subscribe("TopicTest","*");
        //消費事務訊息
        consumer.subscribe("TransactionTopic","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt ext:msgs) {
                    try {
                        System.out.println(new Date() + new String(ext.getBody(),"UTF-8"));
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Start............");
    }
}

重點來了:3.2.6之前的版本這樣寫就可以了，但是之後的版本被關於事務回查這個藉口被閹割了，不會在進行事務回查操作。
那麼第五步向MQ傳送訊息如果失敗的話，會造成A銀行扣款成功而B銀行收款未成功的資料不一致的情況

解決辦法

這裡只需要考慮本地事務執行成功後的情況（因為本地事務失敗不管確認訊息傳送成功與失敗MQ叢集都不會再發送訊息到消費者）：

  1. 本地事務成功後宕機，確認訊息沒有發出，分散式事務只執行一半。
  2. 確認訊息COMMIT_MESSGE發出，但因網路不可達RocketMQ叢集沒收到。
  3. 確認訊息COMMIT_MESSGE發出，RocketMQ叢集收到COMMIT_MESSGE訊息，但rocketmq取消了回查機制，
    生產者還是不知道COMMIT_MESSGE發出是否成功。

上面三種情況的本質是一樣的，就是生產者本地事務成功後，COMMIT_MESSGE訊息是否送達rocketmq叢集；所以可以看做同一種情況.

下面是按key值查詢事務成功提交COMMIT_MESSGE訊息後的返回資訊：

QueryResult 
[indexLastUpdateTimestamp=1516002830440,
    messageList=[
        MessageExt [queueId=1, storeSize=291, queueOffset=0, sysFlag=4, bornTimestamp=1516002831147, bornHost=/192.168.88.1:6313, storeTimestamp=1516002830323, storeHost=/192.168.88.133:10911, msgId=C0A8588500002A9F0000000000000246, commitLogOffset=582, bodyCRC=1229495611, reconsumeTimes=0, preparedTransactionOffset=0, toString()=
            Message 
            [topic=pay, flag=0, properties=
                {KEYS=5cf5dd03-5811-4b7f-b97c-186598e6d08b, TRAN_MSG=true, UNIQ_KEY=C0A8080B2080085EDE7B4B824F2A0000, PGROUP=transaction-pay, TAGS=tag}, 
            body=67]], 

        MessageExt [queueId=1, storeSize=291, queueOffset=0, sysFlag=8, bornTimestamp=1516002831147, bornHost=/192.168.88.1:6313, storeTimestamp=1516002830440, storeHost=/192.168.88.133:10911, msgId=C0A8588500002A9F0000000000000369, commitLogOffset=873, bodyCRC=1229495611, reconsumeTimes=0, preparedTransactionOffset=582, toString()=
            Message 
            [topic=pay, flag=0, properties=
                {KEYS=5cf5dd03-5811-4b7f-b97c-186598e6d08b, TRAN_MSG=true, UNIQ_KEY=C0A8080B2080085EDE7B4B824F2A0000, PGROUP=transaction-pay, TAGS=tag},
            body=67]]
    ]
]

共返回兩條訊息：兩條訊息中大部分資料是一樣的，但sysFlag、storeTimestamp、msgId、commitLogOffset、preparedTransactionOffset欄位是不一樣的：其中第1條為prepared傳送的訊息，第2條只有在提交COMMIT_MESSGE訊息成功後產生。

注意sysFlag、preparedTransactionOffset欄位與prepared訊息的區別，當提交COMMIT_MESSGE訊息成功後，推測MQ叢集做了如下動作：1. 讀取prepared訊息，修改sysFlag、preparedTransactionOffset值，2. 在存入commitlog日誌檔案，設定consumerqueue序列；因為當作一條新的訊息處理，所以toreTimestamp、msgId、commitLogOffset欄位自然也就變了。所以按照發送的prepared訊息的返回結果顯示的msgId檢視sysFlag狀態只是prepared訊息的sysFlag狀態，RocketMQ4.2版本的話要用key值去查詢，才能檢視事務提交成功的訊息標誌sysFlag=8。
下面是按key值查詢事務失敗提交ROLLBACK_MESSAGE訊息後的返回資訊：

QueryResult 
[indexLastUpdateTimestamp=1516002830440,
    messageList=[
        MessageExt [queueId=1, storeSize=291, queueOffset=0, sysFlag=4, bornTimestamp=1516002831147, bornHost=/192.168.88.1:6313, storeTimestamp=1516002830323, storeHost=/192.168.88.133:10911, msgId=C0A8588500002A9F0000000000000246, commitLogOffset=582, bodyCRC=1229495611, reconsumeTimes=0, preparedTransactionOffset=0, toString()=
            Message 
            [topic=pay, flag=0, properties=
                {KEYS=5cf5dd03-5811-4b7f-b97c-186598e6d08b, TRAN_MSG=true, UNIQ_KEY=C0A8080B2080085EDE7B4B824F2A0000, PGROUP=transaction-pay, TAGS=tag}, 
            body=67]]
    ]
]

由此我們可以根據syyFlag判斷我們在提交事務以後, 訊息傳送是否成功

下面是邏輯分析圖

這裡寫圖片描述

一、在執行本地事務commit前向回查表插入訊息的KEY值。

二、在生產者叢集上設定一個定時任務（根據自身分散式事務流程執行的時間設定）。

  1. 從回查表獲取CONFIRM為0的記錄列表，從記錄列表中獲取COUNT為3的記錄，當count列達到指定閥值(假定是3)時：
    此時記錄的COUNT為3，如果CONFIRM還是為0，那麼說明對此事務的回查次數為3，但RocketMQ叢集還未收到COMMIT_MESSAGE訊息，說明發送COMMIT_MESSAGE訊息失敗，但本地事務已經執行成功，那麼必須要重發與此條記錄中KEY值相對應的Perpared訊息的確認訊息。根據KEY值向MQ叢集查詢訊息，根據獲取的訊息重新用同步的方式傳送此條訊息到MQ叢集，並更新此記錄的CONFIRM為1，COUNT+1
  2. 根據第1步獲取的記錄列表，取出CONFIRM為0且COUNT小於3的記錄，根據KEY值向MQ叢集查詢訊息。
  3. 根據第2步獲取的訊息判斷是否是sysFlag為8的訊息；如果是，更新回查表對應KEY記錄的CONFIRM為1，COUNT為count+1，如果不是，更新回查表對應KEY記錄的COUNT為count+1。

這裡是用COUNT來避免剛剛傳送訊息就開始判斷sysFlag的情況,相當於預留了一點rabbitmq的訊息傳送延遲時間,這個也可以使用CREATE_TIME來代替,比如判斷記錄列表中CREATE_TIME在當前時間前3s以前,未成功傳送的訊息

這隻能算其中一種,還有一些手動從生產者和消費者兩端進行資料庫查詢的方法,有興趣的自己可以去了解

相關推薦

RocketMq在閹割訊息回查checkTransactionState後實現分散式事務

利用rocketMQ解決分散式事務 在rocketMQ中生產者有三種角色 NormalProducer（普通）、OrderProducer（順序）、TransactionProducer（事務） 根據名字大概可以看出各個代表著什麼作用，我們這裡用 Trans

[轉載]使用訊息佇列實現分散式事務-公認較為理想的分散式事務解決方案

前陣子從支付寶轉賬1萬塊錢到餘額寶，這是日常生活的一件普通小事，但作為網際網路研發人員的職業病，我就思考支付寶扣除1萬之後，如果系統掛掉怎麼辦，這時餘額寶賬戶並沒有增加1萬，資料就會出現不一致狀況了。 上述場景在各個型別的系統中都能找到相似影子，比如在電商系統中，當有使用者下單後，除了在訂單表插

介紹下用訊息佇列實現分散式事務

    在OIE的時代， 上層應用開發人員總是認為資料庫足夠強大， 所以很多業務可以做的非常簡單。 比如A轉賬50元給B這個過程， 只要寫一個簡單sql語句塊就ok了。         開始事務；     A賬戶減去50     B賬戶增加50     提交事務。     

用訊息系統實現分散式事務

　　從支付寶轉賬1萬塊錢到餘額寶，這是日常生活的一件普通小事，但作為網際網路研發人員的職業病，我就思考支付寶扣除1萬之後，如果系統掛掉怎麼辦，這時餘額寶賬戶並沒有增加1萬，資料就會出現不一致狀況了。 　　上述場景在各個型別的系統中都能找到相似影子，比如在電商系統中，當有使用者下單後，除

使用訊息佇列實現分散式事務-公認較為理想的分散式事務解決方案（三）

  Begin transaction update A set amount=amount-10000 where userId=1; update B set amount=amount+10000 where userId=1; End t

RocketMQ中介軟體實現分散式事務原理分析

案例：Bob向Smith轉賬，那我們到底是先發送訊息，還是先執行扣款操作？  好像都可能會出問題。如果先發訊息，扣款操作失敗，那麼Smith的賬戶裡面會多出一筆錢。反過來，如果先執行扣款操作，後傳送

訊息佇列實現分散式事務

前陣子從支付寶轉賬1萬塊錢到餘額寶，這是日常生活的一件普通小事，但作為網際網路研發人員的職業病，我就思考支付寶扣除1萬之後，如果系統掛掉怎麼辦，這時餘額寶賬戶並沒有增加1萬，資料就會出現不一致狀況了。 上述場景在各個型別的系統中都能找到相似影子，比如在電商系統中，當有使

分散式事務(3)---RocketMQ實現分散式事務原理

分散式事務(3)—RocketMQ實現分散式事務原理 之前講過有關分散式事務2PC、3PC、TCC的理論知識,部落格地址： 1、分散式事務(1)---2PC和3PC原理 2、分散式事務(2）---TCC原理 這篇講有關RocketMQ實現分散式事務的理論知識，下篇也會示例 通過SpringCloud來例項R

分散式事務(4)---RocketMQ實現分散式事務專案

RocketMQ實現分散式事務 有關RocketMQ實現分散式事務前面寫了一篇部落格 1、RocketMQ實現分散式事務原理 下面就這個專案做個整體簡單介紹，並在文字最下方附上專案Github地址。 一、專案概述 1、技術架構 專案總體技術選型 SpringCloud(Finchley.RELEASE) +

關於利用MQ實現分散式事務的想法【轉】

轉自：https://www.jianshu.com/p/bafb09954f18   假設：訊息服務不丟訊息 場景 服務A 服務B 服務C 訊息服務Q 虛擬碼 服務A中 transaction{ A本地事務 B.callB(); C.callC(); A本地事務

springcloud 中使用redis以及rabbitMQ實現分散式事務

分散式事務的補償機制，在很多成熟的案例都是使用TCC機制，來實現資源凍結，以及提交，失敗則釋放資源，很多情況下，處理業務的特殊情況需要對不支援事務的redis進行手動的事務補償。 比如，現在人臉識別入庫的服務，演算法識別人臉的服務，這兩個是依賴性比較高的服務，而且存在長事務，如果需要查詢頻繁

高併發場景系列(一) 利用redis實現分散式事務鎖,解決高併發環境下減庫存

問題描述:某電商平臺,首發一款新品手機,每人限購2臺,預計會有10W的併發,在該情況下,如果扣減庫存,保證不會超賣 方案一 利用資料庫鎖機制,對記錄進行鎖定,再進行操作  SELECT * from goods where ID =1 for updat

使用SQL SERVER的Link Server實現分散式事務

ansi_warning和ansi_nulls 兩個選項必須開啟，好像對系統也沒啥影響。 樓主再檢查一下兩個地方，經測試，在要使用分散式事務時，這是必須要設定的。 1、連結伺服器和名稱解析問題 -- 建立連結伺服器 EXEC sp_addlinkedserver 

利用Rocketmq4.2版來實現分散式事務

花了點時間學了RocketMQ,下面是本人的一點點心得，如果覺的寫的好就點個贊，但如果你要借鑑話，我還是勸你看下面參考資料裡的視訊（作者為阿里牛人），雖然他分享的視訊是為了推銷阿里雲的DRDS、ONS（RocketMQ阿里版），只是講了個大概，沒有細說，但是指明一個大的方向，

springboot搭配TXmanager實現分散式事務

前言：寫這個文章主要是給自己做下筆記以免遺忘，同時也給各大碼農分享一下，互相學習。 分散式事務的需求來源於系統的服務化。在微服務系統中，無法使用傳統的事務達到資料庫的一致性。每個子服務都有自己獨立的資料來源，如果系統初期，沒有分表分庫，每個子服務的資料來源都是連線同一個資料庫主機，同一個資料

利用redis實現分散式事務鎖,解決高併發環境下減庫存

http://download.redis.io/releases/ 安裝： sudo make test 測試編譯 sudo make install 啟動： redis-servre cd “安裝目錄” redis-server ./redis-3.2.9/redis

分析redis訊息佇列和kafka來解決分散式事務場景

1、系統A（扣減托盤）【訊息生產者】 2、系統B（扣減押金）【訊息消費者】 業務描述： 兩套系統，A中扣減托盤，B中對應的要扣減押金；A中托盤歸還，B中押金返還 利用訊息佇列來解決分散式事務過程： 傳送方【生產者】：（不關心接收方狀態，只需要確定本地OK，訊息推送即可）

Java事務--spring+atomikos+tomcat實現分散式事務

        JTA是J2EE的規範之一，如果使用JTA，我們需要去實現相應介面。tomcat是web伺服器，沒有實現J2EE規範，如果我們需要在tomcat伺服器使用JTA，需要自己去實現JTA的介面。本文介紹如何使用tomcat+JTA實現多資料來源的分散式事務。

我是如何基於二階段遞交及悲觀鎖實現分散式事務的

  由於框架一開始的定位就是需要支援強一致性分散式儲存，所以如何實現分散式事務成為一個大挑戰。作者學習了CockroachDB及TiDB等資料庫的實現方式後，決定參考TiDB的實現方式，但不同於使用樂觀方式而是採用悲觀鎖方式，遇到事務衝突採用排隊的方式而不是重啟事務。 一、二階段(2PC)遞交流程： 參考下圖

RocketMQ源碼分析之RocketMQ事務消息實現原理中篇----事務消息狀態回查

下一個 核心 with chcon detail 偏移 ffffff begin n) 上節已經梳理了RocketMQ發送事務消息的流程（基於二階段提交），本節將繼續深入學習事務狀態消息回查，我們知道，第一次提交到消息服務器時消息的主題被替換為RMQ_SYS_TRANS_H