每個(gè)時(shí)代，都不會(huì)虧待會(huì)學(xué)習(xí)的人纺腊。

大家好畔咧，我是 yes。

今天我們來談一談消息隊(duì)列的事務(wù)消息揖膜，一說起事務(wù)相信大家都不陌生誓沸，腦海里蹦出來的就是 ACID。

通常我們理解的事務(wù)就是為了一些更新操作要么都成功壹粟，要么都失敗拜隧，不會(huì)有中間狀態(tài)的產(chǎn)生，而 ACID 是一個(gè)嚴(yán)格的事務(wù)實(shí)現(xiàn)的定義，不過在單體系統(tǒng)時(shí)候一般都不會(huì)嚴(yán)格的遵循 ACID 的約束來實(shí)現(xiàn)事務(wù)洪添，更別說分布式系統(tǒng)了垦页。

分布式系統(tǒng)往往只能妥協(xié)到最終一致性，保證數(shù)據(jù)最終的完整性和一致性薇组，主要原因就是實(shí)力不允許...因?yàn)榭捎眯詾橥酢?/p>

而且要保證完全版的事務(wù)實(shí)現(xiàn)代價(jià)很大外臂，你想想要維護(hù)這么多系統(tǒng)的數(shù)據(jù)坐儿，不允許有中間狀態(tài)數(shù)據(jù)可以被讀取律胀，所有的操作必須不可分割，這意味著一個(gè)事務(wù)的執(zhí)行是阻塞的貌矿，資源是被長(zhǎng)時(shí)間鎖定的炭菌。

在高并發(fā)情況下資源被長(zhǎng)時(shí)間的占用，就是致命的傷害逛漫，舉一個(gè)有味道的例子黑低，如廁高峰期，好了懂得都懂酌毡。

對(duì)了克握， ACID 是什么還不太清楚的同學(xué)，趕緊去查一查枷踏，這里我就不展開說了菩暗。

分布式事務(wù)

那說到分布式事務(wù)，常見的有 2PC旭蠕、TCC 和事務(wù)消息停团，這篇文章重點(diǎn)就是事務(wù)消息，不過 2PC 和 TCC 我稍微提一下掏熬。

2PC

2PC 就是二階段提交佑稠，分別有協(xié)調(diào)者和參與者兩個(gè)角色，二階段分別是準(zhǔn)備階段和提交階段旗芬。

準(zhǔn)備階段就是協(xié)調(diào)者向各參與者發(fā)送準(zhǔn)備命令舌胶，這個(gè)階段參與者除了事務(wù)的提交啥都做了，而提交階段就是協(xié)調(diào)者看看各個(gè)參與者準(zhǔn)備階段都 o 不 ok疮丛，如果有 ok 那么就向各個(gè)參與者發(fā)送提交命令幔嫂，如果有一個(gè)不 ok 那么就發(fā)送回滾命令。

這里的重點(diǎn)就是 2PC 只適用于數(shù)據(jù)庫(kù)層面的事務(wù)这刷，什么意思呢婉烟？就是你想在數(shù)據(jù)庫(kù)里面寫一條數(shù)據(jù)同時(shí)又要上傳一張圖片，這兩個(gè)操作 2PC 無(wú)法保證兩個(gè)操作滿足事務(wù)的約束暇屋。

而且 2PC 是一種強(qiáng)一致性的分布式事務(wù)似袁，它是同步阻塞的，即在接收到提交或回滾命令之前，所有參與者都是互相等待昙衅，特別是執(zhí)行完準(zhǔn)備階段的時(shí)候扬霜，此時(shí)的資源都是鎖定的狀態(tài)，假如有一個(gè)參與者卡了很久而涉，其他參與者都得等它著瓶，產(chǎn)生長(zhǎng)時(shí)間資源鎖定狀態(tài)下的阻塞。

總體而言效率低啼县，并且存在單點(diǎn)故障問題材原，協(xié)調(diào)者是就是那個(gè)單點(diǎn)，并且在極端條件下存在數(shù)據(jù)不一致的風(fēng)險(xiǎn)季眷，例如某個(gè)參與者未收到提交命令余蟹，此時(shí)宕機(jī)了，恢復(fù)之后數(shù)據(jù)是回滾的子刮，而其他參與者其實(shí)都已經(jīng)執(zhí)行了提交事務(wù)的命令了威酒。

TCC

TCC 能保證業(yè)務(wù)層面的事務(wù)，也就是說它不僅僅是數(shù)據(jù)庫(kù)層面挺峡，上面的上傳圖片這種操作它也能做葵孤。

TCC 分為三個(gè)階段 try - confirm - cancel，簡(jiǎn)單的說就是每個(gè)業(yè)務(wù)都需要有這三個(gè)方法橱赠，先都執(zhí)行 try 方法尤仍，這一階段不會(huì)做真正的業(yè)務(wù)操作，只是先占個(gè)坑病线，什么意思呢吓著？比如打算加 10 個(gè)積分，那先在預(yù)添加字段加上這 10 積分送挑，這個(gè)時(shí)候用戶賬上的積分其實(shí)是沒有增加的绑莺。

然后如果都 try 成功了那么就執(zhí)行 confirm 方法，大家都來做真正的業(yè)務(wù)操作惕耕，如果有一個(gè) try 失敗了那么大家都執(zhí)行 cancel 操作纺裁，來撤回剛才的修改。

可以看到 TCC 其實(shí)對(duì)業(yè)務(wù)的耦合性很大司澎，因?yàn)闃I(yè)務(wù)上需要做一定的改造才能完成這三個(gè)方法欺缘，這其實(shí)就是 TCC 的缺點(diǎn)，并且 confirm 和 cancel 操作要注意冪等挤安，因?yàn)榈綀?zhí)行這兩步的時(shí)候沒有退路谚殊，是務(wù)必要完成的，因此需要有重試機(jī)制蛤铜，所以需要保證方法冪等嫩絮。

事務(wù)消息

事務(wù)消息就是今天文章的主角了丛肢，它主要是適用于異步更新的場(chǎng)景，并且對(duì)數(shù)據(jù)實(shí)時(shí)性要求不高的地方剿干。

它的目的是為了解決消息生產(chǎn)者與消息消費(fèi)者的數(shù)據(jù)一致性問題蜂怎。

比如你點(diǎn)外賣，我們先選了炸雞加入購(gòu)物車置尔，又選了瓶可樂杠步，然后下單，付完款這個(gè)流程就結(jié)束了榜轿。

而購(gòu)物車?yán)锩娴臄?shù)據(jù)就很適合用消息通知異步刪除幽歼，因?yàn)橐话愣晕覀兿峦陠尾粫?huì)再去點(diǎn)開這個(gè)店家的菜單，而且就算點(diǎn)開了購(gòu)物車?yán)镞€有這些菜品也沒有關(guān)系差导，影響不大试躏。

我們希望的就是下單成功之后購(gòu)物車的菜品最終會(huì)被刪除猪勇，所以要點(diǎn)就是下單和發(fā)消息這兩個(gè)步驟要么都成功要么都失敗设褐。

RocketMQ 事務(wù)消息

我們先來看一下 RocketMQ 是如何實(shí)現(xiàn)事務(wù)消息的。

RocketMQ 的事務(wù)消息也可以被認(rèn)為是一個(gè)兩階段提交泣刹，簡(jiǎn)單的說就是在事務(wù)開始的時(shí)候會(huì)先發(fā)送一個(gè)半消息給 Broker助析。

半消息的意思就是這個(gè)消息此時(shí)對(duì) Consumer 是不可見的，而且也不是存在真正要發(fā)送的隊(duì)列中椅您，而是一個(gè)特殊隊(duì)列外冀。

發(fā)送完半消息之后再執(zhí)行本地事務(wù)，再根據(jù)本地事務(wù)的執(zhí)行結(jié)果來決定是向 Broker 發(fā)送提交消息掀泳，還是發(fā)送回滾消息雪隧。

此時(shí)有人說這一步發(fā)送提交或者回滾消息失敗了怎么辦？

影響不大员舵，Broker 會(huì)定時(shí)的向 Producer 來反查這個(gè)事務(wù)是否成功脑沿，具體的就是 Producer 需要暴露一個(gè)接口，通過這個(gè)接口 Broker 可以得知事務(wù)到底有沒有執(zhí)行成功马僻，沒成功就返回未知庄拇，因?yàn)橛锌赡苁聞?wù)還在執(zhí)行，會(huì)進(jìn)行多次查詢韭邓。

如果成功那么就將半消息恢復(fù)到正常要發(fā)送的隊(duì)列中措近，這樣消費(fèi)者就可以消費(fèi)這條消息了。

我們?cè)賮砗?jiǎn)單的看下如何使用女淑，我根據(jù)官網(wǎng)示例代碼簡(jiǎn)化了下瞭郑。

可以看到使用起來還是很簡(jiǎn)便直觀的，無(wú)非就是多加個(gè)反查事務(wù)結(jié)果的方法鸭你，然后把本地事務(wù)執(zhí)行的過程寫在 TransationListener 里面屈张。

至此 RocketMQ 事務(wù)消息大致的流程已經(jīng)清晰了我抠，我們畫一張整體的流程圖來過一遍，其實(shí)到第四步這個(gè)消息要么就是正常的消息袜茧，要么就是拋棄什么都不存在菜拓，此時(shí)這個(gè)事務(wù)消息已經(jīng)結(jié)束它的生命周期了。

RocketMQ 事務(wù)消息源碼分析

然后我們?cè)購(gòu)脑创a的角度來看看到底是怎么做的笛厦，首先我們看下sendMessageInTransaction 方法纳鼎，方法有點(diǎn)長(zhǎng)，不過沒有關(guān)系結(jié)構(gòu)還是很清晰的裳凸。

流程也就是我們上面分析的贱鄙，將消息塞入一些屬性，標(biāo)明此時(shí)這個(gè)消息還是半消息姨谷，然后發(fā)送至 Broker逗宁，然后執(zhí)行本地事務(wù)，然后將本地事務(wù)的執(zhí)行狀態(tài)發(fā)送給 Broker 梦湘，我們現(xiàn)在再來看下 Broker 到底是怎么處理這個(gè)消息的瞎颗。

在 Broker 的 SendMessageProcessor#sendMessage 中會(huì)處理這個(gè)半消息請(qǐng)求，因?yàn)榻裉熘饕治龅氖鞘聞?wù)消息捌议，所以其他流程不做分析哼拔，我大致的說一下原理。

簡(jiǎn)單的說就是 sendMessage 中查到接受來的消息的屬性里面MessageConst.PROPERTY_TRANSACTION_PREPARED 是 true 瓣颅，那么可以得知這個(gè)消息是事務(wù)消息倦逐，然后再判斷一下這條消息是否超過最大消費(fèi)次數(shù)，是否要延遲宫补，Broker 是否接受事務(wù)消息等操作后檬姥，將這條消息真正的 topic 和隊(duì)列存入屬性中，然后重置消息的 topic 為RMQ_SYS_TRANS_HALF_TOPIC粉怕，并且隊(duì)列是 0 的隊(duì)列中健民，使得消費(fèi)者無(wú)法讀取這個(gè)消息。

以上就是整體處理半消息的流程斋荞，我們來看一下源碼荞雏。

就是來了波貍貓換太子，其實(shí)延時(shí)消息也是這么實(shí)現(xiàn)的平酿，最終將換了皮的消息入盤凤优。

Broker 處理提交或者回滾消息的處理方法是 EndTransactionProcessor#processRequest，我們來看一看它做了什么操作蜈彼。

可以看到筑辨，如果是提交事務(wù)就是把皮再換回來寫入真正的topic所屬的隊(duì)列中，供消費(fèi)者消費(fèi)幸逆，如果是回滾則是將半消息記錄到一個(gè) half_op 主題下棍辕，到時(shí)候后臺(tái)服務(wù)掃描半消息的時(shí)候就依據(jù)其來判斷這個(gè)消息已經(jīng)處理過了暮现。

那個(gè)后臺(tái)服務(wù)就是 TransactionalMessageCheckService 服務(wù)，它會(huì)定時(shí)的掃描半消息隊(duì)列楚昭，去請(qǐng)求反查接口看看事務(wù)成功了沒栖袋，具體執(zhí)行的就是TransactionalMessageServiceImpl#check 方法。

我大致說一下流程抚太，這一步驟其實(shí)涉及到的代碼很多塘幅，我就不貼代碼了，有興趣的同學(xué)自行了解尿贫。不過我相信用語(yǔ)言也是能說清楚的电媳。

首先取半消息 topic 即RMQ_SYS_TRANS_HALF_TOPIC下的所有隊(duì)列，如果還記得上面內(nèi)容的話庆亡，就知道半消息寫入的隊(duì)列是 id 是 0 的這個(gè)隊(duì)列匾乓，然后取出這個(gè)隊(duì)列對(duì)應(yīng)的 half_op 主題下的隊(duì)列，即 RMQ_SYS_TRANS_OP_HALF_TOPIC 主題下的隊(duì)列又谋。

這個(gè) half_op 主要是為了記錄這個(gè)事務(wù)消息已經(jīng)被處理過拼缝，也就是說已經(jīng)得知此事務(wù)消息是提交的還是回滾的消息會(huì)被記錄在 half_op 中。

然后調(diào)用 fillOpRemoveMap 方法搂根，從 half_op 取一批已經(jīng)處理過的消息來去重珍促，將那些沒有記錄在 half_op 里面的半消息調(diào)用 putBackHalfMsgQueue 又寫入了 commitlog 中，然后發(fā)送事務(wù)反查請(qǐng)求剩愧，這個(gè)反查請(qǐng)求也是 oneWay，即不會(huì)等待響應(yīng)娇斩。當(dāng)然此時(shí)的半消息隊(duì)列的消費(fèi) offset 也會(huì)推進(jìn)仁卷。

然后producer中的 ClientRemotingProcessor#processRequest 會(huì)處理這個(gè)請(qǐng)求，會(huì)把任務(wù)扔到 TransactionMQProducer 的線程池中進(jìn)行犬第，最終會(huì)調(diào)用上面我們發(fā)消息時(shí)候定義的 checkLocalTransactionState 方法锦积，然后將事務(wù)狀態(tài)發(fā)送給 Broker，也是用 oneWay 的方式歉嗓。

看到這里相信大家會(huì)有一些疑問丰介，比如為什么要有個(gè) half_op ，為什么半消息處理了還要再寫入 commitlog 中別急聽我一一道來鉴分。

首先 RocketMQ 的設(shè)計(jì)就是順序追加寫入哮幢，所以說不會(huì)更改已經(jīng)入盤的消息，那事務(wù)消息又需要更新反查的次數(shù)志珍，超過一定反查失敗就判定事務(wù)回滾橙垢。

因此每一次要反查的時(shí)候就將以前的半消息再入盤一次，并且往前推進(jìn)消費(fèi)進(jìn)度伦糯。而 half_op 又會(huì)記錄每一次反查的結(jié)果柜某，不論是提交還是回滾都會(huì)記錄嗽元，因此下一次還循環(huán)到處理此半消息的時(shí)候，可以從 half_op 得知此事務(wù)已經(jīng)結(jié)束了喂击，因此就被過濾掉不需要處理了剂癌。

如果得到的反查的結(jié)果是 UNKNOW，那 half_op 中也不會(huì)記錄此結(jié)果翰绊，因此還能再次反查珍手，并且更新反查次數(shù)。

到現(xiàn)在整個(gè)流程已經(jīng)清晰了辞做，我再畫個(gè)圖總結(jié)一下 Broker 的事務(wù)處理流程琳要。

Kafka 事務(wù)消息

Kafka 的事務(wù)消息和 RocketMQ 的事務(wù)消息又不一樣了，RocketMQ 解決的是本地事務(wù)的執(zhí)行和發(fā)消息這兩個(gè)動(dòng)作滿足事務(wù)的約束秤茅。

而 Kafka 事務(wù)消息則是用在一次事務(wù)中需要發(fā)送多個(gè)消息的情況稚补，保證多個(gè)消息之間的事務(wù)約束，即多條消息要么都發(fā)送成功框喳，要么都發(fā)送失敗课幕，就像下面代碼所演示的。

Kafka 的事務(wù)基本上是配合其冪等機(jī)制來實(shí)現(xiàn) Exactly Once 語(yǔ)義的五垮，所以說 Kafka 的事務(wù)消息不是我們想的那種事務(wù)消息乍惊，RocketMQ 的才是。

講到這我就想扯一下了放仗，說到這個(gè) Exactly Once 其實(shí)不太清楚的同學(xué)很容易會(huì)誤解润绎。

我們知道消息可靠性有三種，分別是最多一次诞挨、恰好一次莉撇、最少一次，之前在消息隊(duì)列連環(huán)問的文章我已經(jīng)提到了基本上我們都是用最少一次然后配合消費(fèi)者端的冪等來實(shí)現(xiàn)恰好一次惶傻。

消息恰好被消費(fèi)一次當(dāng)然我們所有人追求的棍郎，但是之前文章我已經(jīng)從各方面已經(jīng)分析過了峻凫，基本上難以達(dá)到筐付。

而 Kafka 竟說它能實(shí)現(xiàn) Exactly Once扎拣？這么牛啤嗎痪枫？這其實(shí)是 Kafka 的一個(gè)噱頭镐侯，你要說他錯(cuò)撕氧，他還真沒錯(cuò)琼开，你要說他對(duì)但是他實(shí)現(xiàn)的 Exactly Once 不是你心中想的那個(gè) Exactly Once笆呆。

它的恰好一次只能存在一種場(chǎng)景扶认，就是從 Kafka 作為消息源侨拦，然后做了一番操作之后，再寫入 Kafka 中辐宾。

那他是如何實(shí)現(xiàn)恰好一次的狱从？就是通過冪等膨蛮，和我們?cè)跇I(yè)務(wù)上實(shí)現(xiàn)的一樣通過一個(gè)唯一 Id， 然后記錄下來季研，如果已經(jīng)記錄過了就不寫入敞葛，這樣來保證恰好一次。

所以說 Kafka 實(shí)現(xiàn)的是在特定場(chǎng)景下的恰好一次与涡，不是我們所想的利用 Kafka 來發(fā)送消息惹谐，那么這條消息只會(huì)恰巧被消費(fèi)一次。

這其實(shí)和 Redis 說他實(shí)現(xiàn)事務(wù)了一樣驼卖，也不是我們心想的事務(wù)氨肌。

所以開源軟件說啥啥特性開發(fā)出來了，我們一味的相信酌畜，因此其往往都是殘血的或者在特殊的場(chǎng)景下才能滿足怎囚，不要被誤導(dǎo)了，不能相信表面上的描述桥胞，還得詳細(xì)的看看文檔或者源碼恳守。

不過從另一個(gè)角度看也無(wú)可厚非，作為一個(gè)開源軟件肯定是想更多的人用贩虾，我也沒說謊呀催烘，我文檔上寫的很清楚的，這標(biāo)題也沒騙人吧缎罢？

確實(shí)伊群，比如你點(diǎn)進(jìn)震驚xxxx標(biāo)題的文章，人家也沒騙你啥屁使，他自己確實(shí)震驚的呢在岂。

再回來談 Kafka 的事務(wù)消息，所以說這個(gè)事務(wù)消息不是我們想要的那個(gè)事務(wù)消息蛮寂，其實(shí)不是今天的主題了，不過我還是簡(jiǎn)單的說一下易茬。

Kafka 的事務(wù)有事務(wù)協(xié)調(diào)者角色酬蹋，事務(wù)協(xié)調(diào)者其實(shí)就是 Broker 的一部分。

在開始事務(wù)的時(shí)候抽莱，生產(chǎn)者會(huì)向事務(wù)協(xié)調(diào)者發(fā)起請(qǐng)求表示事務(wù)開啟范抓，事務(wù)協(xié)調(diào)者會(huì)將這個(gè)消息記錄到特殊的日志-事務(wù)日志中，然后生產(chǎn)者再發(fā)送真正想要發(fā)送的消息食铐，這里 Kafka 和 RocketMQ 處理不一樣匕垫，Kafka 會(huì)像對(duì)待正常消息一樣處理這些事務(wù)消息，由消費(fèi)端來過濾這個(gè)消息虐呻。

然后發(fā)送完畢之后生產(chǎn)者會(huì)向事務(wù)協(xié)調(diào)者發(fā)送提交或者回滾請(qǐng)求象泵，由事務(wù)協(xié)調(diào)者來進(jìn)行兩階段提交寞秃，如果是提交那么會(huì)先執(zhí)行預(yù)提交，即把事務(wù)的狀態(tài)置為預(yù)提交然后寫入事務(wù)日志偶惠，然后再向所有事務(wù)有關(guān)的分區(qū)寫入一條類似事務(wù)結(jié)束的消息春寿，這樣消費(fèi)端消費(fèi)到這個(gè)消息的時(shí)候就知道事務(wù)好了，可以把消息放出來了忽孽。

最后協(xié)調(diào)者會(huì)向事務(wù)日志中再記一條事務(wù)結(jié)束信息绑改，至此 Kafka 事務(wù)就完成了，我拿 confluent.io 上的圖來總結(jié)一下這個(gè)流程兄一。

最后

至此我們已經(jīng)知道了 RocketMQ 和 Kakfa 的事務(wù)消息全流程厘线，可以看到 RocketMQ 的事務(wù)消息才是我們想要的，當(dāng)然你要是用的流式計(jì)算那么 Kakfa 的事務(wù)消息也是你想要的出革。

需要貼代碼的文章其實(shí)很難受造壮，這貼的多不好，貼的少又怕不清晰蹋盆，真的難费薄，如果覺得文章不錯(cuò)記得點(diǎn)個(gè)在看喲。

我是 yes栖雾，從一點(diǎn)點(diǎn)到億點(diǎn)點(diǎn)楞抡，我們下篇見。