源碼
源代碼: https://github.com/rudolflin/transaction-demo.git
依賴: consul(其實用不到, 只是為了做分布式事務未拆分前的demo使用) ,mysql ,rocketmq
以下部分全部摘自源碼中的readme文件,圖片懶得copy了,建議參考源代碼.
分布式事務--本地消息表, rocketmq
- 采用消息中間件實現, 其實就是保證兩方面, 生產者側本地事務執行成功后一定會發消息,發消息一定會到brocker(代碼實現). 消費者測一定會不斷重試,直到成功(mq實現),極端情況人工處理(代碼實現)
- 人工處理待開發
本地消息表方式
demo a,b,c功能
-
a->b->c三個服務依次調用, 每個服務的account表都減同樣數額的錢
特點
- 最終一致性的前提是認為下游服務一定可以正確完成任務, 如果遇到突發事件, 通過重試最終也能完成.
- 發消息通過本地事務表加定時任務完成,這個是本地消息表完成分布式事務的保證. 保證了本地事務執行一定會發消息,發消息一定會成功.
- 服務消費要實現冪等, 但是要注意處理業務本地數據,寫消息表(向下游發送消息),寫去重表(冪等)要在一個本地事務里面.
- 判斷冪等要在做業務之前,標記成功標志要在業務之后, 盡量和業務分離開.
- 目前想到的冪等方案只有去重表. 其他方案: 布隆過濾器(不是完全准確,有一定的錯誤率).關鍵在於,更新布隆過濾器和或者其他方案的去重標記, 要和業務事務,插入消息表等操作
放在一起.否則可能出現, 業務做完了,做成功的標記沒有打上. - 需要兩張表: task表(記錄要發送mq的任務-生產者使用), 去重表(記錄那些消息處理過-生產者使用)
rocketmq事務方式
-
需要兩張張表: 1、本地事務記錄表, 本地業務執行成功后, 保存一條記錄, 供brocker反查,. 本地業務執行和記錄事務成功狀態要在一個本地事務里. 2、去重表（消費者使用，這種方式消息重復的概率較小）
-
生產者是依靠第5步查詢，保證本地事務執行后，消息一定會發送成功的。 其實就是定時任務部分的工作 ，由mq來做了。
-
事務成功狀態的保存, 要生成一個業務側的key, 用於反查, 這個key可以放在(keys)字段里面 , 這樣在messageExt中就能取到了.
-
事務狀態表, 里邊其實只有一個事務id就可以了, 事務狀態不是必須的 , 因為只要存進來的, 都是成功的, 否則就是失敗了, 也存不進來.
-
service層將參數信息處理后封裝成model . 然后發mq, 這里是沒有事務控制的. 真正的業務和存消息表另起一個service執行. 由mqlistener自動調用執行. 時機是在發送半消息成功之后.
-
transactionid, 直接使用sendmessageintransaction之后,excutelocaltransaction中傳入的message中的id即可.
-
unknow其實只作為調試使用
-
事務回查
-
-
消費者只是普通消費着, 做好冪等即可
-
rocketmq 回查超過一定次數, 就會rollback消息
-
監控報警部分: 需要監控死信隊列 ,或者在消息消費時做記錄, 失敗時存一下記錄, 當失敗多次后出發郵件報警. 因為死信觸發的時間比較長,或者直接改這個死信時間間隔和次數也可以, 正常情況下 , 兩三次失敗后, 下次也會失敗. (兩三次失敗時為了兼容網絡抖動等的影響), 然后做一個消費者組消費死信隊列, 觸發人工報警, 將異常信息放入另一個topic或者數據表里, 等待人工修復.
報警部分待補充
-
rocketmq的回查次數 ,頻率,
-
死信隊列 報警(人工介入), (下游執行失敗)
- 重試隊列是以%RETRY%+consumerGroup作為維度的生成consumeQueue。
- 死信隊列是以%DLQ%+consumerGroup作為維度的生成consumeQueue。
- 進入死信隊列的條件是重試次數超過了最大重試次數。
- 死信隊列的topic是在消息發送過程中判斷對應的topic是否存在，不存在就動態進行創建。