> 每個時代，都不會虧待會學習的人。 大家好，我是 yes。 今天我們來談一談訊息佇列的事務訊息，一說起事務相信大家都不陌生，腦海裡蹦出來的就是 ACID。 通常我們理解的事務就是為了一些更新操作要麼都成功，要麼都失敗，不會有中間狀態的產生，而 ACID 是一個嚴格的事務實現的定義，不過在單體系統時候一般都不會嚴格的遵循 ACID 的約束來實現事務，更別說分散式系統了。 **分散式系統往往只能妥協到最終一致性**，保證資料最終的完整性和一致性，主要原因就是實力不允許...因為可用性為王。 而且要保證完全版的事務實現代價很大，你想想要維護這麼多系統的資料，不允許有中間狀態資料可以被讀取，所有的操作必須不可分割，這意味著一個事務的執行是阻塞的，資源是被長時間鎖定的。 在高併發情況下資源被長時間的佔用，就是致命的傷害，舉一個有味道的例子，如廁高峰期，好了懂得都懂。  對了， ACID 是什麼還不太清楚的同學，趕緊去查一查，這裡我就不展開說了。 ## 分散式事務 那說到分散式事務，常見的有 2PC、TCC 和事務訊息，這篇文章重點就是事務訊息，不過 2PC 和 TCC 我稍微提一下。 ## 2PC 2PC 就是二階段提交，分別有協調者和參與者兩個角色，二階段分別是準備階段和提交階段。 準備階段就是協調者向各參與者傳送準備命令，這個階段參與者除了事務的提交啥都做了，而提交階段就是協調者看看各個參與者準備階段都 o 不 ok，如果有 ok 那麼就向各個參與者傳送提交命令，如果有一個不 ok 那麼就傳送回滾命令。 這裡的重點就是 **2PC 只適用於資料庫層面的事務**，什麼意思呢？就是你想在資料庫裡面寫一條資料同時又要上傳一張圖片，這兩個操作 2PC 無法保證兩個操作滿足事務的約束。 而且 2PC 是一種**強一致性**的分散式事務，它是**同步阻塞**的，即在接收到提交或回滾命令之前，所有參與者都是互相等待，特別是執行完準備階段的時候，此時的資源都是鎖定的狀態，假如有一個參與者卡了很久，其他參與者都得等它，**產生長時間資源鎖定狀態下的阻塞**。 **總體而言效率低**，並且存在**單點故障**問題，協調者是就是那個單點，並且在極端條件下存在**資料不一致**的風險，例如某個參與者未收到提交命令，此時宕機了，恢復之後資料是回滾的，而其他參與者其實都已經執行了提交事務的命令了。 ## TCC **TCC 能保證業務層面的事務**，也就是說它不僅僅是資料庫層面，上面的上傳圖片這種操作它也能做。 TCC 分為三個階段 try - confirm - cancel，簡單的說就是每個業務都需要有這三個方法，先都執行 try 方法，這一階段不會做真正的業務操作，只是先佔個坑，什麼意思呢？比如打算加 10 個積分，那先在預新增欄位加上這 10 積分，這個時候使用者賬上的積分其實是沒有增加的。 然後如果都 try 成功了那麼就執行 confirm 方法，大家都來做真正的業務操作，如果有一個 try 失敗了那麼大家都執行 cancel 操作，來撤回剛才的修改。 可以看到 **TCC 其實對業務的耦合性很大**，因為業務上需要做一定的改造才能完成這三個方法，這其實就是 TCC 的缺點，**並且 confirm 和 cancel 操作要注意冪等**，因為到執行這兩步的時候沒有退路，是務必要完成的，因此需要有重試機制，所以需要保證方法冪等。 ## 事務訊息 事務訊息就是今天文章的主角了，它**主要是適用於非同步更新的場景，並且對資料實時性要求不高的地方**。 它的目的是為了**解決訊息生產者與訊息消費者的資料一致性問題。** 比如你點外賣，我們先選了炸雞加入購物車，又選了瓶可樂，然後下單，付完款這個流程就結束了。  而購物車裡面的資料就很適合用訊息通知非同步刪除，因為一般而言我們下完單不會再去點開這個店家的選單，而且就算點開了購物車裡還有這些菜品也沒有關係，影響不大。 我們希望的就是下單成功之後購物車的菜品最終會被刪除，所以要點就是**下單和發訊息這兩個步驟要麼都成功要麼都失敗**。 ## RocketMQ 事務訊息 我們先來看一下 RocketMQ 是如何實現事務訊息的。 RocketMQ 的事務訊息也可以被認為是一個兩階段提交，簡單的說就是在事務開始的時候會先發送一個半訊息給 Broker。 半訊息的意思就是這個訊息此時對 Consumer 是不可見的，而且也不是存在真正要傳送的佇列中，而是一個特殊佇列。 傳送完半訊息之後再執行本地事務，再根據本地事務的執行結果來決定是向 Broker 傳送提交訊息，還是傳送回滾訊息。 此時有人說這一步傳送提交或者回滾訊息失敗了怎麼辦？ 影響不大，Broker 會定時的向 Producer 來反查這個事務是否成功，具體的就是 Producer 需要暴露一個介面，通過這個介面 Broker 可以得知事務到底有沒有執行成功，沒成功就返回未知，因為有可能事務還在執行，會進行多次查詢。 如果成功那麼就將半訊息恢復到正常要傳送的佇列中，這樣消費者就可以消費這條訊息了。 我們再來簡單的看下如何使用，我根據官網示例程式碼簡化了下。  可以看到使用起來還是很簡便直觀的，無非就是多加個反查事務結果的方法，然後把本地事務執行的過程寫在 TransationListener 裡面。 至此 RocketMQ 事務訊息大致的流程已經清晰了，我們畫一張整體的流程圖來過一遍，其實到第四步這個訊息要麼就是正常的訊息，要麼就是拋棄什麼都不存在，此時這個事務訊息已經結束它的生命週期了。  ## RocketMQ 事務訊息原始碼分析 然後我們再從原始碼的角度來看看到底是怎麼做的，首先我們看下`sendMessageInTransaction` 方法，方法有點長，不過沒有關係結構還是很清晰的。  流程也就是我們上面分析的，將訊息塞入一些屬性，標明此時這個訊息還是半訊息，然後傳送至 Broker，然後執行本地事務，然後將本地事務的執行狀態傳送給 Broker ，我們現在**再來看下 Broker 到底是怎麼處理這個訊息的**。 在 Broker 的 SendMessageProcessor#sendMessage 中會處理這個半訊息請求，因為今天主要分析的是事務訊息，所以其他流程不做分析，我大致的說一下原理。 簡單的說就是 sendMessage 中查到接受來的訊息的屬性裡面`MessageConst.PROPERTY_TRANSACTION_PREPARED` 是 true ，那麼可以得知這個訊息是事務訊息，然後再判斷一下這條訊息是否超過最大消費次數，是否要延遲，Broker 是否接受事務訊息等操作後，將這條訊息真正的 topic 和佇列存入屬性中，然後重置訊息的 topic 為`RMQ_SYS_TRANS_HALF_TOPIC `，並且佇列是 0 的佇列中，使得消費者無法讀取這個訊息。 以上就是整體處理半訊息的流程，我們來看一下原始碼。  就是來了波狸貓換太子，其實延時訊息也是這麼實現的，最終將換了皮的訊息入盤。 Broker 處理提交或者回滾訊息的處理方法是 `EndTransactionProcessor#processRequest`，我們來看一看它做了什麼操作。  可以看到，如果是提交事務就是把皮再換回來寫入真正的topic所屬的佇列中，供消費者消費，如果是回滾則是將半訊息記錄到一個 half_op 主題下，到時候後臺服務掃描半訊息的時候就依據其來判斷這個訊息已經處理過了。 那個後臺服務就是 `TransactionalMessageCheckService` 服務，它會定時的掃描半訊息佇列，去請求反查介面看看事務成功了沒，具體執行的就是`TransactionalMessageServiceImpl#check` 方法。 我大致說一下流程，這一步驟其實涉及到的程式碼很多，我就不貼程式碼了，有興趣的同學自行了解。不過我相信用語言也是能說清楚的。 首先取半訊息 topic 即`RMQ_SYS_TRANS_HALF_TOPIC`下的所有佇列，如果還記得上面內容的話，就知道半訊息寫入的佇列是 id 是 0 的這個佇列，然後取出這個佇列對應的 half_op 主題下的佇列，即 `RMQ_SYS_TRANS_OP_HALF_TOPIC` 主題下的佇列。 這個 half_op 主要是為了記錄這個事務訊息已經被處理過，也就是說已經得知此事務訊息是提交的還是回滾的訊息會被記錄在 half_op 中。 然後呼叫 `fillOpRemoveMap` 方法，從 half_op 取一批已經處理過的訊息來去重，將那些沒有記錄在 half_op 裡面的半訊息呼叫 `putBackHalfMsgQueue` 又寫入了 commitlog 中，然後傳送事務反查請求，這個反查請求也是 oneWay，即不會等待響應。當然此時的半訊息佇列的消費 offset 也會推進。  然後**producer**中的 ClientRemotingProcessor#processRequest 會處理這個請求，會把任務扔到 TransactionMQProducer 的執行緒池中進行，最終會呼叫上面我們發訊息時候定義的 `checkLocalTransactionState` 方法，然後將事務狀態傳送給 Broker，也是用 oneWay 的方式。 看到這裡相信大家會有一些疑問，比如為什麼要有個 half_op ，為什麼半訊息處理了還要再寫入 commitlog 中別急聽我一一道來。 首先 **RocketMQ 的設計就是順序追加寫入，所以說不會更改已經入盤的訊息**，那事務訊息又需要更新反查的次數，超過一定反查失敗就判定事務回滾。 因此每一次要反查的時候就將以前的半訊息再入盤一次，並且往前推進消費進度。而 half_op 又會記錄每一次反查的結果，不論是提交還是回滾都會記錄，因此下一次還迴圈到處理此半訊息的時候，可以從 half_op 得知此事務已經結束了，因此就被過濾掉不需要處理了。 如果得到的反查的結果是 UNKNOW，那 half_op 中也不會記錄此結果，因此還能再次反查，並且更新反查次數。 到現在整個流程已經清晰了，我再畫個圖總結一下 Broker 的事務處理流程。  ## Kafka 事務訊息 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 的事務訊息也是你想要的。 需要貼程式碼的文章其實很難受，這貼的多不好，貼的少又怕不清晰，真的難，如果覺得文章不錯記得點個在看喲。  --- **我是 yes，從一點點到億點點，我們下篇