一、什么是可靠消息最終一致性事務(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)方）和消息中間件之間都是通過(guò)網(wǎng)絡(luò)通信，由于網(wǎng)絡(luò)通信的不確定性會(huì)導(dǎo)致分布式事務(wù)問(wèn)題。

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

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

本地事務(wù)與消息發(fā)送的原子性問(wèn)題即：事務(wù)發(fā)起方在本地事務(wù)執(zhí)行成功后消息必須發(fā)出去，否則就丟棄消息。即實(shí)現(xiàn)本地事務(wù)和消息發(fā)送的原子性，要么都成功，要么都失敗。本地事務(wù)與消息發(fā)送的原子性問(wèn)題是實(shí)現(xiàn)可靠消息最終一致性方案的關(guān)鍵問(wè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;

這種情況下貌似沒(méi)有問(wèn)題，如果發(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èn)題

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

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

二、解決方案

1、本地消息表方案

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

下面以注冊(cè)送積分為例來(lái)說(shuō)明：兩個(gè)微服務(wù)交互，用戶服務(wù)和積分服務(wù)，用戶服務(wù)負(fù)責(zé)添加用戶，積分服務(wù)負(fù)責(zé)增加積分。

交互流程如下：

• 1.用戶注冊(cè)

用戶服務(wù)在本地事務(wù)新增用戶和增加 ”積分消息日志“。（用戶表和消息表通過(guò)本地事務(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)過(guò)第一步消息已經(jīng)寫(xiě)到消息日志表中，可以啟動(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)聽(tīng)MQ，如果消費(fèi)者接收到消息并且業(yè)務(wù)處理完成后向MQ發(fā)送ack（即消息確認(rèn)），此時(shí)說(shuō)明消費(fèi)者正常消費(fèi)消息完成，MQ將不再向消費(fèi)者推送消息，否則消費(fèi)者會(huì)不斷重試向消費(fèi)者來(lái)發(fā)送消息。

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

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

2、RocketMQ事務(wù)消息方案

RocketMQ 是一個(gè)來(lái)自阿里巴巴的分布式消息中間件，于 2012 年開(kāi)源，并在 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í)行的原子性問(wèn)題，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í)行的原子性問(wèn)題。

執(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ù)代碼邏輯，通過(guò)本地?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ù)過(guò)程中，執(zhí)行端掛掉，或者超時(shí)，MQ Server將會(huì)不停的詢問(wèn)同組的其他 Producer來(lái)獲取事務(wù)執(zhí)行狀態(tài)，這個(gè)過(guò)程叫事務(wù)回查。MQ Server會(huì)根據(jù)事務(wù)回查結(jié)果來(lái)決定是否投遞消息。

以上主干流程已由RocketMQ實(shí)現(xiàn)，對(duì)用戶側(cè)來(lái)說(shuō)，用戶需要分別實(shí)現(xiàn)本地事務(wù)執(zhí)行以及本地事務(wù)回查方法，因此只需關(guān)注本地事務(wù)的執(zhí)行狀態(tài)即可，更多關(guān)于java分布式事務(wù)消息一致性的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論