淺談 RocketMQ、Kafka、Pulsar 的事務消息
導語
事務是一個程序執行單元，里面的所有操作要么全部執行成功，要么全部執行失敗。RocketMQ、Kafka 和 Pulsar 都是當今業界應用十分廣泛的開源消息隊列(MQ)組件，筆者在工作中遇到關于 MQ 選型相關的內容，了解到關于“事務消息”這個概念在不同的 MQ 組件里有不同內涵。故借此文，試著淺析一番這三種消息隊列(MQ)的事務消息有何異同，目的是形成關于消息隊列事務消息的全景視圖，給有類似業務需求的同學提供一些參考和借鑒。
一、消息隊列演化
消息隊列(Message Queue，簡稱 MQ)，是指在消息的傳輸中保存消息的容器或服務，是一種異步的服務間通信方式，適用于無服務器和微服務架構，是分布式系統實現高性能、高可用、可伸縮等高級特效的重要組件。常見的主流消息隊列有 ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ、Pulsar 等。而在公司內有 TubeMQ、Ckafka、TDMQ、CMQ、CDMQ、Hippo 等。
消息隊列演化
Kafka：Apache Kafka 是由 Apache 軟件基金會開發的一個開源消息系統項目，由 Scala 寫成。Kafka 最初是由 LinkedIn 開發，并于 2011 年初開源。2012 年 10 月從 Apache Incubator 畢業。該項目的目標是為處理實時數據提供一個統一、高通量、低等待的平臺。Kafka 是一個分布式的、分區的、多復本的日志提交服務。它通過一種獨一無二的設計提供了一個消息系統的功能，其整體架構圖如下所示。
kafka整體架構圖
RocketMQ：Apache RocketMQ 是一個分布式消息和流媒體平臺，具有低延遲、強一致、高性能和可靠性、萬億級容量和靈活的可擴展性。它有借鑒 Kafka 的設計思想，但不是 kafka 的拷貝，其整體架構圖如下所示。
RocketMQ架構圖
Pulsar：Apache Pulsar 是 Apache 軟件基金會頂級項目，是下一代云原生分布式消息流平臺，集消息、存儲、輕量化函數式計算為一體，采用計算與存儲分離架構設計，支持多租戶、持久化存儲、多機房跨區域數據復制，具有強一致性、高吞吐、低延時及高可擴展性等流數據存儲特性，被看作是云原生時代實時消息流傳輸、存儲和計算最佳解決方案，其整體架構圖如下所示。
Pulsar架構圖
二、背景知識
2.1 什么是事務?
2.1.1 事務(Trasaction)
事務是一個程序執行單元，里面的所有操作要么全部執行成功，要么全部執行失敗。
一個事務有四個基本特性，也就是我們常說的(ACID)。
Atomicity(原子性) ：事務是一個不可分割的整體，事務內所有操作要么全做成功，要么全失敗。
Consistency(一致性) ：事務執行前后，數據從一個狀態到另一個狀態必須是一致的(A 向 B 轉賬，不能出現 A 扣了錢，B 卻沒收到)。
Isolation(隔離性)：多個并發事務之間相互隔離，不能互相干擾。
Durablity(持久性) ：事務完成后，對數據的更改是永久保存的，不能回滾。
2.1.2 分布式事務
分布式事務是指事務的參與者、支持事務的服務器、資源服務器以及事務管理器分別位于不同的分布式系統的不同節點之上。分布式事務通常用于在分布式系統中保證不同節點之間的數據一致性。
分布式事務的解決方案一般有以下幾種：
XA(2PC/3PC)
最具有代表性的是由 Oracle Tuxedo 系統提出的 XA 分布式事務協議。XA 中大致分為兩部分：事務管理器和本地資源管理器。其中本地資源管理器往往由數據庫實現，比如 Oracle、DB2 這些商業數據庫都實現了 XA 接口，而事務管理器作為全局的調度者，負責各個本地資源的提交和回滾。XA 協議通常包含**兩階段提交(2PC)和三階段提交(3PC)**兩種實現。兩階段提交顧名思義就是要進行兩個階段的提交：第一階段，準備階段(投票階段) ;第二階段，提交階段(執行階段)。實現過程如下所示：
2PC
二階段提交看似能夠提供原子性的操作，但它存在著一些缺陷，三段提交(3PC)是對兩段提交(2PC)的一種升級優化，有興趣的可以深入了解一下，這里不再贅述。
TCC
TCC(Try-Confirm-Cancel)是 Try、Commit、Cancel 三種指令的縮寫，又被稱補償事務，其邏輯模式類似于 XA 兩階段提交，事務處理流程也很相似，但 2PC 是應用于在 DB 層面，TCC 則可以理解為在應用層面的 2PC，是需要我們編寫業務邏輯來實現。
TCC 它的核心思想是："針對每個操作都要注冊一個與其對應的確認(Try)和補償(Cancel)"。
消息事務
所謂的消息事務就是基于消息隊列的兩階段提交，本質上是對消息隊列的一種特殊利用，它是將本地事務和發消息放在了一個分布式事務里，保證要么本地操作成功成功并且對外發消息成功，要么兩者都失敗。
基于消息隊列的兩階段提交往往用在高并發場景下，將一個分布式事務拆成一個消息事務(A 系統的本地操作+發消息)+B 系統的本地操作，其中 B 系統的操作由消息驅動，只要消息事務成功，那么 A 操作一定成功，消息也一定發出來了，這時候 B 會收到消息去執行本地操作，如果本地操作失敗，消息會重投，直到 B 操作成功，這樣就變相地實現了 A 與 B 的分布式事務。原理如下：
消息事務示意圖
雖然上面的方案能夠完成 A 和 B 的操作，但是 A 和 B 并不是強一致的，而是最終一致(Eventually consistent)的。而這也是滿足 BASE 理論的要求的。這里引申一下，BASE 是 Basically Available(基本可用)、Soft state(軟狀態)和 Eventually consistent (最終一致性)三個短語的縮寫。BASE 理論是對 CAP 中 AP (CAP 已經被證實一個分布式系統最多只能同時滿足 CAP 三項中的兩項)的一個擴展，通過犧牲強一致性來獲得可用性，當出現故障允許部分不可用但要保證核心功能可用，允許數據在一段時間內是不一致的，但最終達到一致狀態。滿足 BASE 理論的事務，我們稱之為“柔性事務”。
2.2 什么是 Exactly-once (精確一次)語義?
在分布式系統中，任何節點都有可能出現異常甚至宕機。在 消息隊列中也一樣，當 Producer 在生產消息時，可能會發生 Broker 宕機不可用，或者網絡突然中斷等異常情況。根據在發生異常時 Producer 處理消息的方式，系統可以具備以下三種消息語義。
2.2.1 At-least-once (至少一次)語義
Producer 通過接收 Broker 的 ACK (消息確認)通知來確保消息成功寫入 Topic。然而，當 Producer 接收 ACK 通知超時，或者收到 Broker 出錯信息時，會嘗試重新發送消息。如果 Broker 正好在成功把消息寫入到 Topic，但還沒有給 Producer 發送 ACK 時宕機，Producer 重新發送的消息會被再次寫入到 Topic，最終導致消息被重復分發至 Consumer。即：消息不會丟失，但有可能被重復發送。
2.2.2 At-most-once (最多一次)語義
當 Producer 在接收 ACK 超時，或者收到 Broker 出錯信息時不重發消息，那就有可能導致這條消息丟失，沒有寫入到 Topic 中，也不會被 Consumer 消費到。在某些場景下，為了避免發生重復消費，我們可以容許消息丟失的發生。即：消息可能會丟失，但絕不會被重復發送。
2.2.3 Exactly-once (精確一次)語義
**Exactly-once 語義保證了即使 Producer 多次發送同一條消息到服務端，服務端也僅僅會記錄一次。**Exactly-once 語義是最可靠的，同時也是最難理解的。Exactly-once 語義需要消息隊列服務端，消息生產端和消費端應用三者的協同才能實現。比如，當消費端應用成功消費并且 ACK 了一條消息之后，又把消費位點回滾到之前的一個消息 ID，那么從那個消息 ID 往后的所有消息都會被消費端應用重新消費到。即：消息不會丟失，也不會被重復發送。
三、RocketMQ、Kafka、Pulsar 事務消息
3.1 RocketMQ 的事務消息
RocketMQ 在 4.3.0 版中已經支持分布式事務消息，這里 RocketMQ 采用了 2PC 的思想來實現了提交事務消息，同時增加一個補償邏輯來處理二階段超時或者失敗的消息，流程如下圖所示：
RocketMQ事務消息
其具體工作流程分為正常事務消息的發送及提交和不正常情況下事務消息的補償流程：
1.在消息隊列上開啟一個事務主題。2.事務中第一個執行的服務發送一條“半消息”(半消息和普通消息的唯一區別是，在事務提交之前，對于消費者來說，這個消息是不可見的)給消息隊列。3.半消息發送成功后，發送半消息的服務就會開始執行本地事務，根據本地事務執行結果來決定事務消息提交或者回滾。
補償流程：RocketMQ 提供事務反查來解決異常情況，如果 RocketMQ 沒有收到提交或者回滾的請求，Broker 會定時到生產者上去反查本地事務的狀態，然后根據生產者本地事務的狀態來處理這個“半消息”是提交還是回滾。值得注意的是我們需要根據自己的業務邏輯來實現反查邏輯接口，然后根據返回值 Broker 決定是提交還是回滾。而且這個反查接口需要是無狀態的，請求到任意一個生產者節點都會返回正確的數據。4.本地事務成功后會讓這個“半消息”變成正常消息，供分布式事務后面的步驟執行自己的本地事務。(這里的事務消息，Producer 不會因為 Consumer 消費失敗而做回滾，采用事務消息的應用，其所追求的是高可用和最終一致性，消息消費失敗的話，RocketMQ 自己會負責重推消息，直到消費成功。)
其中，補償流程用于解決消息 Commit 或者 Rollback 發生超時或者失敗的情況。在 RocketMQ 事務消息的主要流程中，一階段的消息如何對用戶不可見。其中，事務消息相對普通消息最大的特點就是一階段發送的消息對用戶是不可見的。那么，如何做到寫入消息但是對用戶不可見呢?RocketMQ 事務消息的做法是：如果消息是“半消息”，將備份原消息的主題與消息消費隊列，然后改變主題為 RMQ_SYS_TRANS_HALF_TOPIC。由于消費組未訂閱該主題，故消費端無法消費“半消息”的消息，然后 RocketMQ 會開啟一個定時任務，從 Topic 為 RMQ_SYS_TRANS_HALF_TOPIC 中拉取消息進行消費，根據生產者組獲取一個服務提供者發送回查事務狀態請求，根據事務狀態來決定是提交或回滾消息。
講到這里大家就明白了，這里說的就是 2.1.2 節里提到分布式事務中的消息事務，目的是在分布式事務中實現系統的最終一致性。
3.2 Kafka 的事務消息
與 RocketMQ 的事務消息用途不同，Kafka 的事務基本上是配合其冪等機制來實現 Exactly-once (見 2.2.3 節)語義的。
開發此功能的原因可以總結如下。
流處理的需求隨著流處理的興起，對具有更強處理保證的流處理應用的需求也在增長。例如，在金融行業，金融機構使用流處理引擎為用戶處理借款和信貸。這種類型的用例要求每條消息都只處理一次，無一例外。
換句話說，如果流處理應用程序消費消息 A 并將結果作為消息 B (B = f(A))，那么恰好一次處理保證意味著當且僅當 B 被成功生產后 A 才能被標記為消費，反之亦然。
Pulsar事務
事務 API 使流處理應用程序能夠在一個原子操作中使用、處理和生成消息。這意味著，事務中的一批消息可以從許多主題分區接收、生成和確認。一個事務涉及的所有操作都作為整體成功或失敗。
目前，Kafka 默認提供的交付可靠性保障是 At-least-once。如果消息成功“提交”，但 Broker 的應答沒有成功發送回 Producer 端(比如網絡出現瞬時抖動)，那么 Producer 就無法確定消息是否真的提交成功了。因此，它只能選擇重試，這就是 Kafka 默認提供 At-least-once 保障的原因，不過這會導致消息重復發送。大部分用戶還是希望消息只會被交付一次，這樣的話，消息既不會丟失，也不會被重復處理。或者說，即使 Producer 端重復發送了相同的消息，Broker 端也能做到自動去重。在下游 Consumer 看來，消息依然只有一條。那么問題來了，Kafka 是怎么做到精確一次的呢?簡單來說，這是通過兩種機制：冪等性(Idempotence)和事務(Transaction)。
3.2.1 冪等性 Producer
“冪等”這個詞原是數學領域中的概念，指的是某些操作或函數能夠被執行多次，但每次得到的結果都是不變的。冪等性有很多好處，其最大的優勢在于我們可以安全地重試任何冪等性操作，反正它們也不會破壞我們的系統狀態。如果是非冪等性操作，我們還需要擔心某些操作執行多次對狀態的影響，但對于冪等性操作而言，我們根本無需擔心此事。
在 Kafka 中，Producer 默認不是冪等性的，但我們可以創建冪等性 Producer。它其實是 0.11.0.0 版本引入的新功能。enable.idempotence 被設置成 true 后，Producer 自動升級成冪等性 Producer，其他所有的代碼邏輯都不需要改變。Kafka 自動幫你做消息的重復去重。Kafka 為了實現冪等性，它在底層設計架構中引入了ProducerID和SequenceNumber。ProducerID：在每個新的 Producer 初始化時，會被分配一個唯一的 ProducerID，用來標識本次會話。
SequenceNumber：對于每個 ProducerID，Producer 發送數據的每個 Topic 和 Partition 都對應一個從 0 開始單調遞增的 SequenceNumber 值。Broker 在內存維護(pid,seq)映射，收到消息后檢查 seq。Producer 在收到明確的的消息丟失 ack，或者超時后未收到 ack，要進行重試。
new_seq = old_seq+1: 正常消息;
new_seq <= old_seq : 重復消息;
new_seq > old_seq+1: 消息丟失;
另外我們需要了解冪等性 Producer 的作用范圍。首先，它只能保證單分區上的冪等性，即一個冪等性 Producer 能夠保證某個主題的一個分區上不出現重復消息，它無法實現多個分區的冪等性。其次，它只能實現單會話上的冪等性，不能實現跨會話的冪等性。這里的會話，你可以理解為 Producer 進程的一次運行。當你重啟了 Producer 進程之后，這種冪等性保證就喪失了。如果想實現多分區以及多會話上的消息無重復，應該怎么做呢?答案就是事務(transaction)或者依賴事務型 Producer。這也是冪等性 Producer 和事務型 Producer 的最大區別。
3.2.2 事務型 Producer
事務型 Producer 能夠保證將消息原子性地寫入到多個分區中。這批消息要么全部寫入成功，要么全部失敗。另外，事務型 Producer 也不受進程的重啟影響。Producer 重啟后，Kafka 依然保證它們發送消息的 Exactly-once 處理。和普通 Producer 代碼相比，事務型 Producer 的顯著特點是調用了一些事務 API，如 initTransaction、beginTransaction、commitTransaction 和 abortTransaction，它們分別對應事務的初始化、事務開始、事務提交以及事務終止。
Kafka 事務消息是由 Producer、事務協調器、Broker、組協調器、Consumer 等共同參與實現的。
1)Producer
為 Producer 指定固定的 TransactionalId(事務 id)，可以穿越 Producer 的多次會話(Producer 重啟/斷線重連)中，持續標識 Producer 的身份。
每個生產者增加一個 epoch。用于標識同一個 TransactionalId 在一次事務中的 epoch，每次初始化事務時會遞增，從而讓服務端可以知道生產者請求是否舊的請求。使用 epoch 標識 Producer 的每一次"重生"，可以防止同一 Producer 存在多個會話。
Producer 遵從冪等消息的行為，并在發送的 BatchRecord 中增加事務 id 和 epoch。
2)事務協調器(Transaction Coordinator)
引入事務協調器，類似于消費組負載均衡的協調者，每一個實現事務的生產端都被分配到一個事務協調者。以兩階段提交的方式，實現消息的事務提交。
事務協調器使用一個特殊的 Topic：即事務 Topic，事務 Topic 本身也是持久化的，日志信息記錄事務狀態信息，由事務協調者寫入。
事務協調器通過 RPC 調用，協調 Broker 和 Consumer 實現事務的兩階段提交。
每一個 Broker 都會啟動一個事務協調器，使用 hash(TransactionalId)確定 Producer 對應的事務協調器，使得整個集群的負載均衡。
3)Broker
引入控制消息(Control Messages)：這些消息是客戶端產生的并寫入到主題的特殊消息，但對于使用者來說不可見。它們是用來讓 Broker 告知消費者之前拉取的消息是否被原子性提交。
Broker 處理事務協調器的 commit/abort 控制消息，把控制消息向正常消息一樣寫入 Topic(圖中標 c 的消息，和正常消息交織在一起，用來確認事務提交的日志偏移)，并向前推進消息提交偏移 hw。
kafka事務
4)組協調器
如果在事務過程中，提交了消費偏移，組協調器在 offset log 中寫入事務消費偏移。當事務提交時，在 offset log 中寫入事務 offset 確認消息。
5)Consumer
Consumer 過濾未提交消息和事務控制消息，使這些消息對用戶不可見。
有兩種實現方式，
- Consumer 緩存方式
設置 isolation.level=read_uncommitted，此時 topic 的所有消息對 Consumer 都可見。Consumer 緩存這些消息，直到收到事務控制消息。若事務 commit，則對外發布這些消息;若事務 abort，則丟棄這些消息。
- Broker 過濾方式
設置 isolation.level=read_committed，此時 topic 中未提交的消息對 Consumer 不可見，只有在事務結束后，消息才對 Consumer 可見。Broker 給 Consumer 的 BatchRecord 消息中，會包含以列表，指明哪些是"abort"事務，Consumer 丟棄 abort 事務的消息即可。
因為事務機制會影響消費者所能看到的消息的范圍，它不只是簡單依賴高水位來判斷。它依靠一個名為 LSO(Log Stable Offset)的位移值來判斷事務型消費者的可見性。
3.3 Pulsar 的事務消息
Apache Pulsar 在 2.8.0 正式支持了事務相關的功能，Pulsar 這里提供的事務區別于 RocketMQ 中 2PC 那種事務的實現方式，沒有本地事務回查的機制，更類似于 Kafka 的事務實現機制。Apache Pulsar 中的事務主要用來保證類似 Pulsar Functions 這種流計算場景中 Exactly-once 語義的實現，這也符合 Apache Pulsar 本身 Event Streaming 的定位，即保證端到端(End-to-End)的事務實現的語義。
在 Pulsar 中，對于事務語義是這樣定義的：允許事件流應用將消費、處理、生產消息整個過程定義為一個原子操作，即生產者或消費者能夠處理跨多個主題和分區的消息，并確保這些消息作為一個單元被處理。
Pulsar 事務具有以下語義：
- 事務中的所有操作都作為一個單元提交。要么提交所有消息，要么都不提交。
- 每條消息只寫入或處理一次，不會丟失數據或重復(即使發生故障)。
- 如果事務中止，則此事務中的所有寫入和確認都將回滾。
事務中的批量消息可以被以多分區接收、生產和確認。
- 消費者只能讀取已提交(確認)的消息。換句話說，Broker 不傳遞屬于打開事務的事務消息或屬于中止事務的消息。
- 跨多個分區的消息寫入是原子性的。
- 跨多個訂閱的消息確認是原子性的。訂閱下的消費者在確認帶有事務 ID 的消息時，只會成功確認一次消息。
Pulsar 事務消息由以下幾個關鍵點構成：
1)事務 ID
事務 ID(TxnID)標識 Pulsar 中的唯一事務。事務 ID 長度是 128-bit。最高 16 位保留給事務協調器的 ID，其余位用于每個事務協調器中單調遞增的數字。
2)事務協調器(Transaction Coordinator)
事務協調器(TC)是運行在 Pulsar Broker 中的一個模塊。
- 它維護事務的整個生命周期，并防止事務進入錯誤狀態。
- 它處理事務超時，并確保事務在事務超時后中止。
3)事務日志
所有事務元數據都保存在事務日志中。事務日志由 Pulsar 主題記錄。如果事務協調器崩潰，它可以從事務日志恢復事務元數據。
事務日志存儲事務狀態，而不是事務中的實際消息(實際消息存儲在實際的主題分區中)。
4)事務緩存
向事務內的主題分區生成的消息存儲在該主題分區的事務緩沖區(TB)中。在提交事務之前，事務緩沖區中的消息對消費者不可見。當事務中止時，事務緩沖區中的消息將被丟棄。
事務緩沖區將所有正在進行和中止的事務存儲在內存中。所有消息都發送到實際的分區 Pulsar 主題。提交事務后，事務緩沖區中的消息對消費者具體化(可見)。事務中止時，事務緩沖區中的消息將被丟棄。
5)待確認狀態
掛起確認狀態在事務完成之前維護事務中的消息確認。如果消息處于掛起確認狀態，則在該消息從掛起確認狀態中移除之前，其他事務無法確認該消息。
掛起的確認狀態被保留到掛起的確認日志中(cursor ledger)。新啟動的 broker 可以從掛起的確認日志中恢復狀態，以確保狀態確認不會丟失。
處理流程一般分為以下幾個步驟：
- 開啟事務。
- 使用事務發布消息。
- 使用事務確認消息。
- 結束事務。
Pulsar 的事務處理流程與 Kafka 的事務處理思路大致上保持一致，大家都有一個 TC 以及對應的一個用于持久化 TC 所有操作的 Topic 來記錄所有事務狀態變更的請求。同樣的在事務開始階段也都有一個專門的 Topic 來去 查詢 TC 對應的 Owner Broker 的位置在哪里。不同的是，第一：Kafka 中對于未確認的消息是維護在 Broker 端的，但是 Pulsar 的是維護在 Client 端的，通過 Transaction Timeout 來決定這個事務是否執行成功，所以有了 Transaction Timeout 的存在之后，就可以確保 Client 和 Broker 側事務處理的一致性。第二：由于 Kafka 本身沒有單條消息的 Ack，所以 Kafka 的事務處理只能是順序執行的，當一個事務請求被阻塞之后，會阻塞后續所有的事務請求，但是 Pulsar 是可以對消息進行單條 Ack 的，所以在這里每一個事務的 Ack 動作是獨立的，不會出現事務阻塞的情況。
四、結論
RocketMQ 和 Kafka/Pulsar 的事務消息實用的場景是不一樣的。
RocketMQ 中的事務，它解決的問題是，確保執行本地事務和發消息這兩個操作，要么都成功，要么都失敗。并且 RocketMQ 增加了一個事務反查的機制，來盡量提高事務執行的成功率和數據一致性。
Kafka 中的事務，它解決的問題是，確保在一個事務中發送的多條消息，要么都成功，要么都失敗。(這里面的多條消息不一定要在同一個主題和分區中，可以是發往多個主題和分區的消息)當然也可以在 kafka 事務執行過程中開啟本地事務來實現類似 RocketMQ 事務消息的效果，但是 Kafka 是沒有事務消息反查機制的，它是直接拋出異常的，用戶可以根據異常來實現自己的重試等方法保證事務正常運行。
它們的共同點就是：都是通過兩階段提交來實現事務的，事務消息都保存在單獨的主題上。不同的地方就是 RocketMQ 是通過“半消息”來實現的，kafka 是直接將消息發送給對應的 topic，通過客戶端來過濾實現的。而且它們兩個使用的場景區別是非常之大的，RockteMQ 主要解決的是基于本地事務和消息的數據一致性，而 Kafka 的事務則是用于實現它的 Exactly-once 機制，應用于實時流計算的場景中。
Pulsar 的事務消息和 Kafka 應用場景和語義類似，只是由于底層實現機制有差別，在一些細節上有區別。
相信看到這里就非常清楚了，對于事務消息如何選型和應用，首先要明白你的業務需求是什么。是要實現分布式事務的最終一致性，還是要實現 Exactly-once (精確一次)語義?明白之后需求，選擇什么組件就十分明確了。
參考文章
pulsar 官方 doc
消息隊列漫談：如何使用消息隊列實現分布式事務?
