每個時代，都不會虧待會學習的人。
今天我們來談一談消息隊列的事務消息，一說起事務相信大家都不陌生，腦海裡蹦出來的就是 ACID。
通常我們了解的事務就是為了一些更新操作要麼都成功，要麼都失敗，不會有中間狀态的産生，而 ACID 是一個嚴格的事務實作的定義，不過在單體系統時候一般都不會嚴格的遵循 ACID 的限制來實作事務，更别說分布式系統了。
分布式系統往往隻能妥協到最終一緻性，保證資料最終的完整性和一緻性，主要原因就是實力不允許...因為可用性為王。
而且要保證完全版的事務實作代價很大，你想想要維護這麼多系統的資料，不允許有中間狀态資料可以被讀取，所有的操作必須不可分割，這意味着一個事務的執行是阻塞的，資源是被長時間鎖定的。
在高并發情況下資源被長時間的占用，就是緻命的傷害，舉一個有味道的例子，如廁高峰期，好了懂得都懂。
對了， ACID 是什麼還不太清楚的同學，趕緊去查一查，這裡我就不展開說了。
那說到分布式事務，常見的有 2PC、TCC 和事務消息，這篇文章重點就是事務消息，不過 2PC 和 TCC 我稍微提一下。
2PC 就是二階段送出，分别有協調者和參與者兩個角色，二階段分别是準備階段和送出階段。
準備階段就是協調者向各參與者發送準備指令，這個階段參與者除了事務的送出啥都做了，而送出階段就是協調者看看各個參與者準備階段都 o 不 ok，如果有 ok 那麼就向各個參與者發送送出指令，如果有一個不 ok 那麼就發送復原指令。
這裡的重點就是 2PC 隻适用于資料庫層面的事務，什麼意思呢？就是你想在資料庫裡面寫一條資料同時又要上傳一張圖檔，這兩個操作 2PC 無法保證兩個操作滿足事務的限制。
而且 2PC 是一種強一緻性的分布式事務，它是同步阻塞的，即在接收到送出或復原指令之前，所有參與者都是互相等待，特别是執行完準備階段的時候，此時的資源都是鎖定的狀态，假如有一個參與者卡了很久，其他參與者都得等它，産生長時間資源鎖定狀态下的阻塞。
總體而言效率低，并且存在單點故障問題，協調者是就是那個單點，并且在極端條件下存在資料不一緻的風險，例如某個參與者未收到送出指令，此時當機了，恢複之後資料是復原的，而其他參與者其實都已經執行了送出事務的指令了。
TCC 能保證業務層面的事務，也就是說它不僅僅是資料庫層面，上面的上傳圖檔這種操作它也能做。
TCC 分為三個階段 try - confirm - cancel，簡單的說就是每個業務都需要有這三個方法，先都執行 try 方法，這一階段不會做真正的業務操作，隻是先占個坑，什麼意思呢？比如打算加 10 個積分，那先在預添加字段加上這 10 積分，這個時候使用者賬上的積分其實是沒有增加的。
然後如果都 try 成功了那麼就執行 confirm 方法，大家都來做真正的業務操作，如果有一個 try 失敗了那麼大家都執行 cancel 操作，來撤回剛才的修改。
可以看到 TCC 其實對業務的耦合性很大，因為業務上需要做一定的改造才能完成這三個方法，這其實就是 TCC 的缺點，并且 confirm 和 cancel 操作要注意幂等，因為到執行這兩步的時候沒有退路，是務必要完成的，是以需要有重試機制，是以需要保證方法幂等。
事務消息就是今天文章的主角了，它主要是适用于異步更新的場景，并且對資料實時性要求不高的地方。
它的目的是為了解決消息生産者與消息消費者的資料一緻性問題。
比如你點外賣，我們先選了炸雞加入購物車，又選了瓶可樂，然後下單，付完款這個流程就結束了。
而購物車裡面的資料就很适合用消息通知異步删除，因為一般而言我們下完單不會再去點開這個店家的菜單，而且就算點開了購物車裡還有這些菜品也沒有關系，影響不大。
我們希望的就是下單成功之後購物車的菜品最終會被删除，是以要點就是下單和發消息這兩個步驟要麼都成功要麼都失敗。
我們先來看一下 RocketMQ 是如何實作事務消息的。
RocketMQ 的事務消息也可以被認為是一個兩階段送出，簡單的說就是在事務開始的時候會先發送一個半消息給 Broker。
半消息的意思就是這個消息此時對 Consumer 是不可見的，而且也不是存在真正要發送的隊列中，而是一個特殊隊列。
發送完半消息之後再執行本地事務，再根據本地事務的執行結果來決定是向 Broker 發送送出消息，還是發送復原消息。
此時有人說這一步發送送出或者復原消息失敗了怎麼辦？
影響不大，Broker 會定時的向 Producer 來反查這個事務是否成功，具體的就是 Producer 需要暴露一個接口，通過這個接口 Broker 可以得知事務到底有沒有執行成功，沒成功就傳回未知，因為有可能事務還在執行，會進行多次查詢。
如果成功那麼就将半消息恢複到正常要發送的隊列中，這樣消費者就可以消費這條消息了。
我們再來簡單的看下如何使用，我根據官網示例代碼簡化了下。
可以看到使用起來還是很簡便直覺的，無非就是多加個反查事務結果的方法，然後把本地事務執行的過程寫在 TransationListener 裡面。
至此 RocketMQ 事務消息大緻的流程已經清晰了，我們畫一張整體的流程圖來過一遍，其實到第四步這個消息要麼就是正常的消息，要麼就是抛棄什麼都不存在，此時這個事務消息已經結束它的生命周期了。
然後我們再從源碼的角度來看看到底是怎麼做的，首先我們看下<code>sendMessageInTransaction</code> 方法，方法有點長，不過沒有關系結構還是很清晰的。
流程也就是我們上面分析的，将消息塞入一些屬性，标明此時這個消息還是半消息，然後發送至 Broker，然後執行本地事務，然後将本地事務的執行狀态發送給 Broker ，我們現在再來看下 Broker 到底是怎麼處理這個消息的。
在 Broker 的 SendMessageProcessor#sendMessage 中會處理這個半消息請求，因為今天主要分析的是事務消息，是以其他流程不做分析，我大緻的說一下原理。
簡單的說就是 sendMessage 中查到接受來的消息的屬性裡面<code>MessageConst.PROPERTY_TRANSACTION_PREPARED</code> 是 true ，那麼可以得知這個消息是事務消息，然後再判斷一下這條消息是否超過最大消費次數，是否要延遲，Broker 是否接受事務消息等操作後，将這條消息真正的 topic 和隊列存入屬性中，然後重置消息的 topic 為<code>RMQ_SYS_TRANS_HALF_TOPIC</code>，并且隊列是 0 的隊列中，使得消費者無法讀取這個消息。
以上就是整體處理半消息的流程，我們來看一下源碼。
就是來了波狸貓換太子，其實延時消息也是這麼實作的，最終将換了皮的消息入盤。
Broker 處理送出或者復原消息的處理方法是 <code>EndTransactionProcessor#processRequest</code>，我們來看一看它做了什麼操作。
可以看到，如果是送出事務就是把皮再換回來寫入真正的topic所屬的隊列中，供消費者消費，如果是復原則是将半消息記錄到一個 half_op 主題下，到時候背景服務掃描半消息的時候就依據其來判斷這個消息已經處理過了。
那個背景服務就是 <code>TransactionalMessageCheckService</code> 服務，它會定時的掃描半消息隊列，去請求反查接口看看事務成功了沒，具體執行的就是<code>TransactionalMessageServiceImpl#check</code> 方法。
我大緻說一下流程，這一步驟其實涉及到的代碼很多，我就不貼代碼了，有興趣的同學自行了解。不過我相信用語言也是能說清楚的。
首先取半消息 topic 即<code>RMQ_SYS_TRANS_HALF_TOPIC</code>下的所有隊列，如果還記得上面内容的話，就知道半消息寫入的隊列是 id 是 0 的這個隊列，然後取出這個隊列對應的 half_op 主題下的隊列，即 <code>RMQ_SYS_TRANS_OP_HALF_TOPIC</code> 主題下的隊列。
這個 half_op 主要是為了記錄這個事務消息已經被處理過，也就是說已經得知此事務消息是送出的還是復原的消息會被記錄在 half_op 中。
然後調用 <code>fillOpRemoveMap</code> 方法，從 half_op 取一批已經處理過的消息來去重，将那些沒有記錄在 half_op 裡面的半消息調用 <code>putBackHalfMsgQueue</code> 又寫入了 commitlog 中，然後發送事務反查請求，這個反查請求也是 oneWay，即不會等待響應。當然此時的半消息隊列的消費 offset 也會推進。
然後producer中的 ClientRemotingProcessor#processRequest 會處理這個請求，會把任務扔到 TransactionMQProducer 的線程池中進行，最終會調用上面我們發消息時候定義的 <code>checkLocalTransactionState</code> 方法，然後将事務狀态發送給 Broker，也是用 oneWay 的方式。
看到這裡相信大家會有一些疑問，比如為什麼要有個 half_op ，為什麼半消息處理了還要再寫入 commitlog 中别急聽我一一道來。
首先 RocketMQ 的設計就是順序追加寫入，是以說不會更改已經入盤的消息，那事務消息又需要更新反查的次數，超過一定反查失敗就判定事務復原。
是以每一次要反查的時候就将以前的半消息再入盤一次，并且往前推進消費進度。而 half_op 又會記錄每一次反查的結果，不論是送出還是復原都會記錄，是以下一次還循環到處理此半消息的時候，可以從 half_op 得知此事務已經結束了，是以就被過濾掉不需要處理了。
如果得到的反查的結果是 UNKNOW，那 half_op 中也不會記錄此結果，是以還能再次反查，并且更新反查次數。
到現在整個流程已經清晰了，我再畫個圖總結一下 Broker 的事務處理流程。
Kafka 的事務消息和 RocketMQ 的事務消息又不一樣了，RocketMQ 解決的是本地事務的執行和發消息這兩個動作滿足事務的限制。
而 Kafka 事務消息則是用在一次事務中需要發送多個消息的情況，保證多個消息之間的事務限制，即多條消息要麼都發送成功，要麼都發送失敗，就像下面代碼所示範的。
Kafka 的事務基本上是配合其幂等機制來實作 Exactly Once 語義的，是以說 Kafka 的事務消息不是我們想的那種事務消息，RocketMQ 的才是。
講到這我就想扯一下了，說到這個 Exactly Once 其實不太清楚的同學很容易會誤解。
我們知道消息可靠性有三種，分别是最多一次、恰好一次、最少一次，之前在消息隊列連環問的文章我已經提到了基本上我們都是用最少一次然後配合消費者端的幂等來實作恰好一次。
消息恰好被消費一次當然我們所有人追求的，但是之前文章我已經從各方面已經分析過了，基本上難以達到。
而 Kafka 竟說它能實作 Exactly Once？這麼牛啤嗎？這其實是 Kafka 的一個噱頭，你要說他錯，他還真沒錯，你要說他對但是他實作的 Exactly Once 不是你心中想的那個 Exactly Once。
它的恰好一次隻能存在一種場景，就是從 Kafka 作為消息源，然後做了一番操作之後，再寫入 Kafka 中。
那他是如何實作恰好一次的？就是通過幂等，和我們在業務上實作的一樣通過一個唯一 Id， 然後記錄下來，如果已經記錄過了就不寫入，這樣來保證恰好一次。
是以說 Kafka 實作的是在特定場景下的恰好一次，不是我們所想的利用 Kafka 來發送消息，那麼這條消息隻會恰巧被消費一次。
這其實和 Redis 說他實作事務了一樣，也不是我們心想的事務。
是以開源軟體說啥啥特性開發出來了，我們一味的相信，是以其往往都是殘血的或者在特殊的場景下才能滿足，不要被誤導了，不能相信表面上的描述，還得詳細的看看文檔或者源碼。
不過從另一個角度看也無可厚非，作為一個開源軟體肯定是想更多的人用，我也沒說謊呀，我文檔上寫的很清楚的，這标題也沒騙人吧？
确實，比如你點進震驚xxxx标題的文章，人家也沒騙你啥，他自己确實震驚的呢。
再回來談 Kafka 的事務消息，是以說這個事務消息不是我們想要的那個事務消息，其實不是今天的主題了，不過我還是簡單的說一下。
Kafka 的事務有事務協調者角色，事務協調者其實就是 Broker 的一部分。
在開始事務的時候，生産者會向事務協調者發起請求表示事務開啟，事務協調者會将這個消息記錄到特殊的日志-事務日志中，然後生産者再發送真正想要發送的消息，這裡 Kafka 和 RocketMQ 處理不一樣，Kafka 會像對待正常消息一樣處理這些事務消息，由消費端來過濾這個消息。
然後發送完畢之後生産者會向事務協調者發送送出或者復原請求，由事務協調者來進行兩階段送出，如果是送出那麼會先執行預送出，即把事務的狀态置為預送出然後寫入事務日志，然後再向所有事務有關的分區寫入一條類似事務結束的消息，這樣消費端消費到這個消息的時候就知道事務好了，可以把消息放出來了。
最後協調者會向事務日志中再記一條事務結束資訊，至此 Kafka 事務就完成了，我拿 confluent.io 上的圖來總結一下這個流程。
至此我們已經知道了 RocketMQ 和 Kakfa 的事務消息全流程，可以看到 RocketMQ 的事務消息才是我們想要的，當然你要是用的流式計算那麼 Kakfa 的事務消息也是你想要的。
需要貼代碼的文章其實很難受，這貼的多不好，貼的少又怕不清晰，真的難，如果覺得文章不錯記得點個在看喲。