1.什么是可靠消息最終一致性事務(wù)

? 可靠消息最終一致性方案是指當(dāng)事務(wù)發(fā)起方執(zhí)行完成本地事務(wù)后并發(fā)出一條消息，事務(wù)參與方(消息消費(fèi)者)一定能夠接收消息并處理事務(wù)成功秸妥，此方案強(qiáng)調(diào)的是只要消息發(fā)給事務(wù)參與方最終事務(wù)要達(dá)到一致颊咬。

? 此方案是利用消息中間件完成抗蠢，如下圖：

? 事務(wù)發(fā)起方（消息生產(chǎn)方）將消息發(fā)給消息中間件薛躬，事務(wù)參與方從消息中間件接收消息，事務(wù)發(fā)起方和消息中間件之間僧界，事務(wù)參與方（消息消費(fèi)方）和消息中間件之間都是通過網(wǎng)絡(luò)通信侨嘀，由于網(wǎng)絡(luò)通信的不確定性會(huì)導(dǎo)致分布式事務(wù)問題。

因此可靠消息最終一致性方案要解決以下幾個(gè)問題：

1）.本地事務(wù)與消息發(fā)送的原子性問題

? 本地事務(wù)與消息發(fā)送的原子性問題即：事務(wù)發(fā)起方在本地事務(wù)執(zhí)行成功后消息必須發(fā)出去捂襟，否則就丟棄消息咬腕。即實(shí)現(xiàn)本地事務(wù)和消息發(fā)送的原子性，要么都成功葬荷，要么都失敗涨共。本地事務(wù)與消息發(fā)送的原子性問題是實(shí)現(xiàn)可靠消息最終一致性方案的關(guān)鍵問題。

先來(lái)嘗試下這種操作宠漩，先發(fā)送消息举反，再操作數(shù)據(jù)庫(kù)：

begin transaction；

//1.發(fā)送MQ

//2.數(shù)據(jù)庫(kù)操作

commit transation;

這種情況下無(wú)法保證數(shù)據(jù)庫(kù)操作與發(fā)送消息的一致性扒吁，因?yàn)榭赡馨l(fā)送消息成功照筑，數(shù)據(jù)庫(kù)操作失敗。

你立馬想到第二種方案，先進(jìn)行數(shù)據(jù)庫(kù)操作凝危，再發(fā)送消息：

begin transaction；

//1.數(shù)據(jù)庫(kù)操作

//2.發(fā)送MQ

commit transation;

? 這種情況下貌似沒有問題晨逝，如果發(fā)送MQ消息失敗蛾默，就會(huì)拋出異常，導(dǎo)致數(shù)據(jù)庫(kù)事務(wù)回滾捉貌。但如果是超時(shí)異常支鸡，數(shù)據(jù)庫(kù)回滾，但MQ其實(shí)已經(jīng)正常發(fā)送了趁窃，同樣會(huì)導(dǎo)致不一致牧挣。

2）事務(wù)參與方接收消息的可靠性

事務(wù)參與方必須能夠從消息隊(duì)列接收到消息，如果接收消息失敗可以重復(fù)接收消息醒陆。

3）消息重復(fù)消費(fèi)的問題

? 由于網(wǎng)絡(luò)2的存在瀑构，若某一個(gè)消費(fèi)節(jié)點(diǎn)超時(shí)但是消費(fèi)成功，此時(shí)消息中間件會(huì)重復(fù)投遞此消息刨摩，就導(dǎo)致了消息的重復(fù)消費(fèi)寺晌。

? 要解決消息重復(fù)消費(fèi)的問題就要實(shí)現(xiàn)事務(wù)參與方的方法冪等性。

2.解決方案

? 上節(jié)討論了可靠消息最終一致性事務(wù)方案需要解決的問題澡刹，本節(jié)討論具體的解決方案呻征。

2.1.本地消息表方案

? 本地消息表這個(gè)方案最初是eBay提出的，此方案的核心是通過本地事務(wù)保證數(shù)據(jù)業(yè)務(wù)操作和消息的一致性罢浇，然后通過定時(shí)任務(wù)將消息發(fā)送至消息中間件陆赋，待確認(rèn)消息發(fā)送給消費(fèi)方成功再將消息刪除栈虚。

下面以注冊(cè)送積分為例來(lái)說明：

下例共有兩個(gè)微服務(wù)交互讥此，用戶服務(wù)和積分服務(wù)，用戶服務(wù)負(fù)責(zé)添加用戶症副，積分服務(wù)負(fù)責(zé)增加積分凌受。

交互流程如下：

1）用戶注冊(cè)

? 用戶服務(wù)在本地事務(wù)新增用戶和增加 ”積分消息日志“阵子。（用戶表和消息表通過本地事務(wù)保證一致）

下邊是偽代碼

begin transaction；

//1.新增用戶

//2.存儲(chǔ)積分消息日志

commit transation;

這種情況下胜蛉，本地?cái)?shù)據(jù)庫(kù)操作與存儲(chǔ)積分消息日志處于同一個(gè)事務(wù)中挠进，本地?cái)?shù)據(jù)庫(kù)操作與記錄消息日志操作具備原子性。

2）定時(shí)任務(wù)掃描日志

? 如何保證將消息發(fā)送給消息隊(duì)列呢誊册？

? 經(jīng)過第一步消息已經(jīng)寫到消息日志表中领突，可以啟動(dòng)獨(dú)立的線程，定時(shí)對(duì)消息日志表中的消息進(jìn)行掃描并發(fā)送至消息中間件案怯，在消息中間件反饋發(fā)送成功后刪除該消息日志君旦，否則等待定時(shí)任務(wù)下一周期重試。

3）消費(fèi)消息

? 如何保證消費(fèi)者一定能消費(fèi)到消息呢？

? 這里可以使用MQ的ack（即消息確認(rèn)）機(jī)制金砍，消費(fèi)者監(jiān)聽MQ局蚀，如果消費(fèi)者接收到消息并且業(yè)務(wù)處理完成后向MQ發(fā)送ack（即消息確認(rèn)），此時(shí)說明消費(fèi)者正常消費(fèi)消息完成恕稠，MQ將不再向消費(fèi)者推送消息琅绅，否則消費(fèi)者會(huì)不斷重試向消費(fèi)者來(lái)發(fā)送消息。

? 積分服務(wù)接收到”增加積分“消息鹅巍，開始增加積分千扶，積分增加成功后向消息中間件回應(yīng)ack，否則消息中間件將重復(fù)投遞此消息骆捧。

? 由于消息會(huì)重復(fù)投遞澎羞，積分服務(wù)的”增加積分“功能需要實(shí)現(xiàn)冪等性。

2.2.RocketMQ事務(wù)消息方案

? RocketMQ 是一個(gè)來(lái)自阿里巴巴的分布式消息中間件敛苇，于 2012 年開源妆绞，并在 2017 年正式成為 Apache 頂級(jí)項(xiàng)目。據(jù)了解接谨，包括阿里云上的消息產(chǎn)品以及收購(gòu)的子公司在內(nèi)摆碉，阿里集團(tuán)的消息產(chǎn)品全線都運(yùn)行在 RocketMQ 之上，并且最近幾年的雙十一大促中脓豪，RocketMQ 都有搶眼表現(xiàn)巷帝。Apache RocketMQ 4.3之后的版本正式支持事務(wù)消息，為分布式事務(wù)實(shí)現(xiàn)提供了便利性支持扫夜。

? RocketMQ 事務(wù)消息設(shè)計(jì)則主要是為了解決 Producer 端的消息發(fā)送與本地事務(wù)執(zhí)行的原子性問題楞泼，RocketMQ 的設(shè)計(jì)中 broker 與 producer 端的雙向通信能力，使得 broker 天生可以作為一個(gè)事務(wù)協(xié)調(diào)者存在笤闯；而 RocketMQ 本身提供的存儲(chǔ)機(jī)制為事務(wù)消息提供了持久化能力堕阔；RocketMQ 的高可用機(jī)制以及可靠消息設(shè)計(jì)則為事務(wù)消息在系統(tǒng)發(fā)生異常時(shí)依然能夠保證達(dá)成事務(wù)的最終一致性。

? 在RocketMQ 4.3后實(shí)現(xiàn)了完整的事務(wù)消息颗味，實(shí)際上其實(shí)是對(duì)本地消息表的一個(gè)封裝超陆，將本地消息表移動(dòng)到了MQ內(nèi)部，解決 Producer 端的消息發(fā)送與本地事務(wù)執(zhí)行的原子性問題浦马。

執(zhí)行流程如下：

為方便理解我們還以注冊(cè)送積分的例子來(lái)描述 整個(gè)流程时呀。

Producer 即MQ發(fā)送方，本例中是用戶服務(wù)晶默，負(fù)責(zé)新增用戶谨娜。MQ訂閱方即消息消費(fèi)方，本例中是積分服務(wù)磺陡，負(fù)責(zé)新增積分趴梢。

1漠畜、Producer 發(fā)送事務(wù)消息

? Producer （MQ發(fā)送方）發(fā)送事務(wù)消息至MQ Server，MQ Server將消息狀態(tài)標(biāo)記為Prepared（預(yù)備狀態(tài)）坞靶，注意此時(shí)這條消息消費(fèi)者（MQ訂閱方）是無(wú)法消費(fèi)到的憔狞。

? 本例中，Producer 發(fā)送 ”增加積分消息“ 到MQ Server滩愁。

2躯喇、MQ Server回應(yīng)消息發(fā)送成功

? MQ Server接收到Producer 發(fā)送給的消息則回應(yīng)發(fā)送成功表示MQ已接收到消息。

3硝枉、Producer 執(zhí)行本地事務(wù)

? Producer 端執(zhí)行業(yè)務(wù)代碼邏輯，通過本地?cái)?shù)據(jù)庫(kù)事務(wù)控制倦微。

? 本例中妻味，Producer 執(zhí)行添加用戶操作。

4欣福、消息投遞

? 若Producer 本地事務(wù)執(zhí)行成功則自動(dòng)向MQServer發(fā)送commit消息责球，MQ Server接收到commit消息后將”增加積分消息“ 狀態(tài)標(biāo)記為可消費(fèi)，此時(shí)MQ訂閱方（積分服務(wù)）即正常消費(fèi)消息拓劝；

? 若Producer 本地事務(wù)執(zhí)行失敗則自動(dòng)向MQServer發(fā)送rollback消息雏逾，MQ Server接收到rollback消息后 將刪除”增加積分消息“ 。

? MQ訂閱方（積分服務(wù)）消費(fèi)消息郑临，消費(fèi)成功則向MQ回應(yīng)ack栖博，否則將重復(fù)接收消息。這里ack默認(rèn)自動(dòng)回應(yīng)厢洞，即程序執(zhí)行正常則自動(dòng)回應(yīng)ack仇让。

5、事務(wù)回查

? 如果執(zhí)行Producer端本地事務(wù)過程中躺翻，執(zhí)行端掛掉丧叽，或者超時(shí)，MQ Server將會(huì)不停的詢問同組的其他 Producer來(lái)獲取事務(wù)執(zhí)行狀態(tài)公你，這個(gè)過程叫事務(wù)回查踊淳。MQ Server會(huì)根據(jù)事務(wù)回查結(jié)果來(lái)決定是否投遞消息。

以上主干流程已由RocketMQ實(shí)現(xiàn)陕靠，對(duì)用戶側(cè)來(lái)說迂尝，用戶需要分別實(shí)現(xiàn)本地事務(wù)執(zhí)行以及本地事務(wù)回查方法，因此只需關(guān)注本地事務(wù)的執(zhí)行狀態(tài)即可懦傍。

RoacketMQ提供RocketMQLocalTransactionListener接口：

public interface RocketMQLocalTransactionListener {

? /**

? - 發(fā)送prepare消息成功此方法被回調(diào)雹舀，該方法用于執(zhí)行本地事務(wù)

? - @param msg 回傳的消息，利用transactionId即可獲取到該消息的唯一Id

? - @param arg 調(diào)用send方法時(shí)傳遞的參數(shù)粗俱，當(dāng)send時(shí)候若有額外的參數(shù)可以傳遞到send方法中说榆，這里能獲取到

? - @return 返回事務(wù)狀態(tài)虚吟，COMMIT：提交? ROLLBACK：回滾? UNKNOW：回調(diào)

? ? */

? ? ? RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg);

? /**

? - @param msg 通過獲取transactionId來(lái)判斷這條消息的本地事務(wù)執(zhí)行狀態(tài)

? - @return 返回事務(wù)狀態(tài)，COMMIT：提交? ROLLBACK：回滾? UNKNOW：回調(diào)

? ? */

? ? ? RocketMQLocalTransactionState checkLocalTransaction(Message msg);

? }

發(fā)送事務(wù)消息：

以下是RocketMQ提供用于發(fā)送事務(wù)消息的API：

TransactionMQProducer producer = new TransactionMQProducer("ProducerGroup");

producer.setNamesrvAddr("127.0.0.1:9876");

producer.start();

//設(shè)置TransactionListener實(shí)現(xiàn)

producer.setTransactionListener(transactionListener）签财；

//發(fā)送事務(wù)消息

SendResult sendResult = producer.sendMessageInTransaction(msg, null);

3 RocketMQ實(shí)現(xiàn)可靠消息案例

本實(shí)例通過RocketMQ中間件實(shí)現(xiàn)可靠消息最終一致性分布式事務(wù)串慰，模擬兩個(gè)賬戶的轉(zhuǎn)賬交易過程。

案例內(nèi)容參見：http://www.pbteach.com/post/java_distribut/subject_dtx-04/

本案例使用了RocketMQ作為消息中間件唱蒸，RocketMQ主要解決了兩個(gè)功能：

? 1邦鲫、本地事務(wù)與消息發(fā)送的原子性問題。

? 2神汹、事務(wù)參與方接收消息的可靠性庆捺。

4 總結(jié)

? 可靠消息最終一致性事務(wù)適合執(zhí)行周期長(zhǎng)且實(shí)時(shí)性要求不高的場(chǎng)景。引入消息機(jī)制后屁魏，同步的事務(wù)操作變?yōu)榛谙?zhí)行的異步操作, 避免了分布式事務(wù)中的同步阻塞操作的影響滔以，并實(shí)現(xiàn)了兩個(gè)服務(wù)的解耦。