spring cloud實現可靠消息一致性

1.前言

1.1 目的

1. 爲開發測試提供指導性文件
2. 爲系統從此的擴展提供參考
3. 解決系統中消息不可達問題

1.2 範圍和功能

1.3 適用讀者

• 須要發送MQ分佈式系統的開發人員和測試人員
• 可靠消息服務的開發人員和測試人員

1.4 讀者須知

本服務須要提供一個sdk和數據庫初始語句建立數據庫表,而且對外提供可掃描的domain、mapper、service,使用的技術框架zk + mapper3 + pagehelper + feign(edas) , 使用者(上游系統、下游系統) 只須要在對應的接口上寫上響應註解便可實現可靠消息, 若是不熟悉上述框架,可選擇對應框架替換,好比redis替換zk,放棄mapper3和pagehelper使用傳統的mybatis,使用http接口替換fein(eads)的解決辦法,本文不提供替換的解決方案

1.5 參考文檔

2 系統概述

本文爲分佈式系統解決方案,此方案涉及 3 個模塊：

1. 上游應用，執行業務併發送指令給可靠消息服務並保留消息副本。
2. 可靠消息服務和 MQ消息組件，協調上下游消息的傳遞，並確保上下游數據的一致性。
3. 下游應用，監聽 MQ 的消息並執行自身業務並保留消息副本。

2.1業務流程圖

2.2數據庫表設計

2.2.1 可靠消息表

2.2.2 消費者確認表

2.2.3 消費者表

2.2.4 生產者表、

2.2.5 發佈關係表

2.2.6 消息重發記錄表

暫時未設計

2.2.7 消息訂閱關係表

2.2.8 消息訂閱TAG關係表

2.2.9 各個子系統消息落地的消息表

推薦一個交流學習羣：478030634 裏面會分享一些資深架構師錄製的視頻錄像：有Spring，MyBatis，Netty源碼分析，高併發、高性能、分佈式、微服務架構的原理，JVM性能優化這些成爲架構師必備的知識體系。還能領取免費的學習資源，目前受益良多

3 詳細設計

3.1 上游應用執行業務併發送 MQ 消息

上游應用將本地業務執行和消息發送綁定在同一個本地事務中，保證要麼本地操做成功併發送 MQ 消息，要麼兩步操做都失敗並回滾。這裏能夠採用自定義切面完成,後續會有介紹。

1. 上游應用發送待確認消息到可靠消息系統。(本地消息落地)
2. 可靠消息系統保存待確認消息並返回。
3. 上游應用執行本地業務。
4. 上游應用通知可靠消息系統確認業務已執行併發送消息。
5. 可靠消息系統修改消息狀態爲發送狀態並將消息投遞到 MQ 中間件。

以上每一步均可能出現失敗狀況，分析一下這 5 步出現異常後上遊業務和消息發送是否一致：

上游應用執行完成，下游應用還沒有執行或執行失敗時，此事務即處於 BASE 理論的 Soft State 狀態。

3.2 下游應用監聽 MQ 消息並執行業務

1. 下游應用監聽 MQ 消息並執行業務，而且將消息的消費結果通知可靠消息服務。(本地消息落地)
2. 可靠消息的狀態須要和下游應用的業務執行保持一致，可靠消息狀態不是已完成時，確保下游應用未執行，可靠消息狀態是已完成時，確保下游應用已執行。

下游應用和可靠消息服務之間的交互圖以下：

1. 下游應用監聽 MQ 消息組件並獲取消息, 並存儲本地消息

4.下游系統通知可靠消息服務已接收到消息

5.可靠消息把消息更新爲已接收狀態

6.下游應用根據 MQ 消息體信息處理本地業務

7.下游應用向 MQ 組件自動發送 ACK 確認消息被消費

8.下游應用通知可靠消息系統消息被成功消費，可靠消息將該消息狀態更改成以消費,任務表狀態修改成已完成。

1. 下游應用監聽 MQ 消息組件並獲取消息, 並存儲本地消息
2. 下游系統通知可靠消息服務已接收到消息
3. 可靠消息把消息更新爲已接收狀態
4. 下游應用根據 MQ 消息體信息處理本地業務
5. 下游應用向 MQ 組件自動發送 ACK 確認消息被消費
6. 下游應用通知可靠消息系統消息被成功消費，可靠消息將該消息狀態更改成已消費,任務表狀態修改成已完成

3.3 生產者消息狀態確認

可靠消息服務定時監聽消息的狀態，若是存在狀態爲待確認而且超時的消息，則表示上游應用和可靠消息交互中的步驟 4 或者 5 出現異常。

可靠消息則攜帶消息體內的信息向上遊應用發起請求查詢該業務是否已執行。上游應用提供一個可查詢接口供可靠消息追溯業務執行狀態，若是業務執行成功則更改消息狀態爲已發送，不然刪除此消息確保數據一致。具體流程以下:

3.4 消費者消息狀態確認

下游消費MQ服務異步通知可靠消息的過程當中可能出現異常,在此可能致使兩個現象1、消息已接到但可靠消息沒有確認接到2、消息已成功消費但可靠消息沒有確認接到,爲此下游系統須要提供消費者消息狀態查詢接口,從而可靠消息從新確認.在確認過程當中若是是可靠消息爲已消費而下游消費系統爲已接收則不進行更新操做. 具體流程以下：

3.5 消息重投

消息已發送則表示上游應用已經執行，接下來則確保下游應用也能正常執行。 可靠消息服務發現可靠消息服務中存在消息狀態爲已發送而且超時的消息，則表示可靠消息服務和下游應用中存在異常的步驟，不管哪一個步驟出現異常，可靠消息服務都將此消息從新投遞到 MQ 組件中供下游應用監聽。 下游應用監聽到此消息後，在保證冪等性的狀況下從新執行業務並通知可靠消息服務此消息已經成功消費，最終確保上游應用、下游應用的數據最終一致性。具體流程以下：

1. 可靠消息服務定時查詢狀態爲已發送並超時的消息
2. 可靠消息將消息從新投遞到 MQ 組件中
3. 下游應用監聽消息，在知足冪等性的條件下，從新執行業務。
4. 下游應用通知可靠消息服務該消息已經成功消費。
5. 更新consumer消息記錄爲已消費

3.6 刪除上游系統7天前成功發送的消息

在預發送執行MQ消息的時候本地消息若是落庫則須要刪除消息,不然業務系統須要額外提供查詢消息發送狀態接口, 這裏介紹兩種方法

第一種,RPC服務接口來實現, 在生產者和消費者註冊到可靠消息的時候把生產者和消費者存儲到BeanFactory的Map裏在定時清理任務的時候去處理在線的RPC服務

第二種,發可靠消息來實現, 確保100%到達

3.7 刪除下游系統7天前成功消費的消息

在消費MQ消息的時候本地消息若是落庫則須要刪除消息,不然業務系統須要額外提供查詢消息發送狀態接口,刪除實現同3.6

3.8 天天備份可靠消息記錄

天天將成功消息刪除並備份到對應數據庫提供歷史消息查詢功能,固然若是你選擇mongo能夠不考慮備份消息

推薦一個交流學習羣：478030634 裏面會分享一些資深架構師錄製的視頻錄像：有Spring，MyBatis，Netty源碼分析，高併發、高性能、分佈式、微服務架構的原理，JVM性能優化這些成爲架構師必備的知識體系。還能領取免費的學習資源，目前受益良多

4 核心代碼實現

這裏作一個說明，由於項目採用的是rocketmq，一個topic對應一個生產者，而可靠消息採用的是中間件負責發送消息，又不能採用中間件的生產者爲全部上游系統發送消息，這裏引入了zookeeper作註冊中心，因此依賴可靠消息的服務，在啓動項目的時候會像中間件去註冊生產者，而中間件的watch機制會及時的更新生產者和消費者狀態，而中間件會爲使用中間件的系統提供sdk，使用者無需關注實現，只須要引入中間件的sdk和對應的註解便可完成可靠消息的發送和消費，詳見下圖： 普通消息發送流程：
可靠消息發送流程：

可靠消息發送和消費：

流程：

服務註冊

消費註解 @MqProducerStore

生產註解@MqConsumerStore

定時清理全部訂閱者消費成功的消息數據

定時清理全部生產者發送成功的消息數據

定時清理全部生產者發送成功的消息數據

處理髮送中的消息數據

處理待確認的消息數據

可靠消息用法

例子

強制： 須要使用的使用加上述兩個註解，方法參數須要加入 MqMessageData

若是對本文感興趣，或者本文對您有所幫助，可靠參考github代碼，本套代碼是spring cloud E版本 + vue兩套全家桶實現

若是有時間最好能給點加個星或者follow一下，筆者在這裏先謝過了。對不知道怎麼加星的朋友，請用電腦登陸github或者碼雲,這裏兩個截圖

寫在最後

更多內容請參考paascloud ： https://document.paascloud.net/