1. 前言

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

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

2. MQ的配置建議

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

主從同步策略

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

刷盤策略

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

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

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

3.1 消息持久化

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

• • 消息id

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

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

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

• • 分區(qū)

• • 消息體

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

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

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

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

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

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

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

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

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

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

3.2 消息補(bǔ)償設(shè)計(jì)

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

補(bǔ)償代碼

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

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

4. 消費(fèi)方設(shè)計(jì)

消費(fèi)方相對比較簡單，主要有兩點(diǎn)要求

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

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

以下是消費(fèi)表的設(shè)計(jì)

• • 消息id

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

• • Topic

• • 分區(qū)

• • 偏移量

• • 消息體

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

• • 異常信息（消費(fèi)失敗會記錄異常信息）

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

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

消費(fèi)方代碼

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

圖7 消費(fèi)方部分核心代碼

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

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

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