1. 前言

業(yè)務(wù)系統(tǒng)之間通過MQ進行交互時，怎么保證發(fā)送的消息對方一定能收到，可能有人說RocketMQ就能做到，如果貴公司用到的消息隊列是kafka、rabbitmq、activemq實現(xiàn)

這里分享一下基于消息的分布式事務(wù)解決方案，此種方案是最終一致性的解決方案，不挑MQ，但是前提MQ本身要支持接收到的消息不能丟失。

2. MQ的配置建議

如果要保證MQ接收到的消息不丟，就要配置相關(guān)的同步策略或者刷盤策略

主從同步策略

建議主從同步建議設(shè)置為主從同步策略為主從同步完再響應(yīng)，這樣單個節(jié)點如果掛了，另一個節(jié)點的數(shù)據(jù)還會存在

刷盤策略

消息中間件為了提高效率，默認接收到消息不會立即刷盤，如果要主從同步策略是主節(jié)點接收到消息以后立即響應(yīng)，這會正好主節(jié)點宕機，就會導(dǎo)致消息丟失，所以要特別注意下，雖然可以設(shè)置成同步刷盤，但是效率就會降低，所以還是建議設(shè)置主從同步策略

3. 生產(chǎn)方設(shè)計

生產(chǎn)者的職責(zé)是必須要保證本地事務(wù)提交成功消息一定要發(fā)送出去，或者業(yè)務(wù)處理失敗就不發(fā)送。

3.1 消息持久化

生產(chǎn)方方案如下，首先需要在業(yè)務(wù)庫中創(chuàng)建一張表，字段大致為：

• • 消息id

• • 業(yè)務(wù)id

• • 業(yè)務(wù)方名稱（如果一個庫是多個子系統(tǒng)在用就需要這個字段）

• • topic（發(fā)送消息主題）

• • 分區(qū)

• • 消息體

• • 狀態(tài)（0未發(fā)送 1已發(fā)送）

• • 是否失敗重試（ 0不重試 1重試）

與本地業(yè)務(wù)表使用同一個事務(wù)，提交則一起提交，回滾則一起回滾，因為使用的同一個事務(wù)所以是強一致的，再事務(wù)提交以后進行消息數(shù)據(jù)的發(fā)送，發(fā)送成功以后則更改消息狀態(tài)為已發(fā)送，具體流程請查看圖1

   
圖1 消息正常發(fā)送流程

這里可能還有一點還要考慮，就是在圖1的第2步、第3步、第4步會出現(xiàn)失敗，具體描述如下：

1. 1. 如插入本地庫成功，但是發(fā)送MQ失敗

2. 2. 消息發(fā)送成功，但是響應(yīng)失敗，比如超時，其實這會MQ已經(jīng)接收發(fā)送方的消息了，但是發(fā)送方不清楚

3. 3. 消息發(fā)送響應(yīng)都成功了，但是更改本地表狀態(tài)為已發(fā)送失敗了。

持久化相關(guān)代碼

圖4，集成Spring的事務(wù)管理器,重寫事務(wù)提交后發(fā)送消息

3.2 消息補償設(shè)計

以上這三個問題就需要引入補償任務(wù)來處理了，具體查看圖5，補償任務(wù)會根據(jù)發(fā)送狀態(tài)查詢對應(yīng)的數(shù)據(jù)，然后進行發(fā)送，這里有一點特別注意，消費方要必須做冪等處理，因為圖1的第3步、第4步消息都已經(jīng)發(fā)送到MQ了，只是發(fā)送方不清楚，所以還會重復(fù)發(fā)送，另外99.9%的場景是能立即發(fā)送成功的，只有很小部分需要做補償：

補償代碼

查詢待發(fā)送的數(shù)據(jù)，這里為1分鐘之前的，定時任務(wù)用的是elastic-job，用其他定時任務(wù)也可以

至此整個發(fā)送方設(shè)計就完成了，下面看看部分

4. 消費方設(shè)計

消費方相對比較簡單，主要有兩點要求

1. 1. 保證消息不會重復(fù)消費

2. 2. 記錄消息便于消息對賬，對賬主要是極端情況下，那些消息沒收到，便于重新投遞

以下是消費表的設(shè)計

• • 消息id

• • 業(yè)務(wù)id

• • Topic

• • 分區(qū)

• • 偏移量

• • 消息體

• • 狀態(tài)（0未消費成功 1消費成功 2消費失敗 ）

• • 異常信息（消費失敗會記錄異常信息）

• • 業(yè)務(wù)方名稱（如果一個庫是多個子系統(tǒng)在用就需要這個字段）

此表也要與業(yè)務(wù)表處于同一個事務(wù)，如果不是一個事務(wù)，會出現(xiàn)業(yè)務(wù)表操作成功、消息表插入失敗，如果出現(xiàn)消息重復(fù)發(fā)送就會出現(xiàn)重復(fù)消費的問題，具體查看圖6

消費方代碼

這里是kafka的消費代碼，通過動態(tài)代理，封裝KafkaListener類，在處理前進行消息重復(fù)判斷，在處理后進行消費表的插入，這里需要特別注意一點，業(yè)務(wù)處理不能把異常自己吃掉，否則上層捕獲不到，會認為業(yè)務(wù)處理成功，從而插入臟數(shù)據(jù)

圖7 消費方部分核心代碼

5. 歷史數(shù)據(jù)清理

通過前面介紹，我們創(chuàng)建了2張表，分別為消息發(fā)送表、消息消費表，這兩張表要特別注意下，如果業(yè)務(wù)量比較大，數(shù)據(jù)量會快速增長，所以需要刪除已經(jīng)處理成功的數(shù)據(jù)，通過配置兩個定時任務(wù)，保留一定的時間數(shù)據(jù)，其他時間的數(shù)據(jù)就可以刪除了，代碼如下：

圖9 消費方清理數(shù)據(jù)代碼