新同事上來就用Kafka，瑟瑟發(fā)抖...

作者：劉若愚 2021-09-16 07:55:39

開發(fā)

架構(gòu)

開發(fā)工具

Kafka 事務(wù)是一個程序執(zhí)行單元，里面的所有操作要么全部執(zhí)行成功，要么全部執(zhí)行失敗。

在清鎮(zhèn)等地區(qū)，都構(gòu)建了全面的區(qū)域性戰(zhàn)略布局，加強(qiáng)發(fā)展的系統(tǒng)性、市場前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務(wù)理念，為客戶提供成都網(wǎng)站制作、成都網(wǎng)站設(shè)計 網(wǎng)站設(shè)計制作按需網(wǎng)站制作,公司網(wǎng)站建設(shè),企業(yè)網(wǎng)站建設(shè),品牌網(wǎng)站建設(shè),營銷型網(wǎng)站,外貿(mào)網(wǎng)站建設(shè),清鎮(zhèn)網(wǎng)站建設(shè)費(fèi)用合理。

圖片來自 包圖網(wǎng)

RocketMQ、Kafka 和 Pulsar 都是當(dāng)今業(yè)界應(yīng)用十分廣泛的開源消息隊(duì)列(MQ)組件。

筆者在工作中遇到關(guān)于 MQ 選型相關(guān)的內(nèi)容，了解到關(guān)于“事務(wù)消息”這個概念在不同的 MQ 組件里有不同內(nèi)涵。

故借此文，試著淺析一番這三種消息隊(duì)列(MQ)的事務(wù)消息有何異同，目的是形成關(guān)于消息隊(duì)列事務(wù)消息的全景視圖，給有類似業(yè)務(wù)需求的同學(xué)提供一些參考和借鑒。

消息隊(duì)列演化

消息隊(duì)列(Message Queue，簡稱 MQ)，是指在消息的傳輸中保存消息的容器或服務(wù)，是一種異步的服務(wù)間通信方式，適用于無服務(wù)器和微服務(wù)架構(gòu)，是分布式系統(tǒng)實(shí)現(xiàn)高性能、高可用、可伸縮等高級特效的重要組件。

常見的主流消息隊(duì)列有 ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ、Pulsar 等。

而在公司內(nèi)有 TubeMQ、Ckafka、TDMQ、CMQ、CDMQ、Hippo 等。

①Kafka

Apache Kafka 是由 Apache 軟件基金會開發(fā)的一個開源消息系統(tǒng)項(xiàng)目，由 Scala 寫成。

Kafka 最初是由 LinkedIn 開發(fā)，并于 2011 年初開源。2012 年 10 月從 Apache Incubator 畢業(yè)。該項(xiàng)目的目標(biāo)是為處理實(shí)時數(shù)據(jù)提供一個統(tǒng)一、高通量、低等待的平臺。

Kafka 是一個分布式的、分區(qū)的、多復(fù)本的日志提交服務(wù)。它通過一種獨(dú)一無二的設(shè)計提供了一個消息系統(tǒng)的功能。

其整體架構(gòu)圖如下所示：

②RocketMQ

Apache RocketMQ 是一個分布式消息和流媒體平臺，具有低延遲、強(qiáng)一致、高性能和可靠性、萬億級容量和靈活的可擴(kuò)展性。它有借鑒 Kafka 的設(shè)計思想，但不是 Kafka 的拷貝。

其整體架構(gòu)圖如下所示：

③Pulsar

Apache Pulsar 是 Apache 軟件基金會頂級項(xiàng)目，是下一代云原生分布式消息流平臺。

它集消息、存儲、輕量化函數(shù)式計算為一體，采用計算與存儲分離架構(gòu)設(shè)計，支持多租戶、持久化存儲、多機(jī)房跨區(qū)域數(shù)據(jù)復(fù)制，具有強(qiáng)一致性、高吞吐、低延時及高可擴(kuò)展性等流數(shù)據(jù)存儲特性，被看作是云原生時代實(shí)時消息流傳輸、存儲和計算最佳解決方案。

其整體架構(gòu)圖如下所示：

背景知識

①什么是事務(wù)?

事務(wù)：是一個程序執(zhí)行單元，里面的所有操作要么全部執(zhí)行成功，要么全部執(zhí)行失敗。

一個事務(wù)有四個基本特性，也就是我們常說的(ACID)：

• Atomicity(原子性)：事務(wù)是一個不可分割的整體，事務(wù)內(nèi)所有操作要么全做成功，要么全失敗。
• Consistency(一致性)：事務(wù)執(zhí)行前后，數(shù)據(jù)從一個狀態(tài)到另一個狀態(tài)必須是一致的(A 向 B 轉(zhuǎn)賬，不能出現(xiàn) A 扣了錢，B 卻沒收到)。
• Isolation(隔離性)：多個并發(fā)事務(wù)之間相互隔離，不能互相干擾。
• Durablity(持久性)：事務(wù)完成后，對數(shù)據(jù)的更改是永久保存的，不能回滾。

分布式事務(wù)：是指事務(wù)的參與者、支持事務(wù)的服務(wù)器、資源服務(wù)器以及事務(wù)管理器分別位于不同的分布式系統(tǒng)的不同節(jié)點(diǎn)之上。分布式事務(wù)通常用于在分布式系統(tǒng)中保證不同節(jié)點(diǎn)之間的數(shù)據(jù)一致性。

分布式事務(wù)的解決方案一般有以下幾種：

XA(2PC/3PC)：最具有代表性的是由 Oracle Tuxedo 系統(tǒng)提出的 XA 分布式事務(wù)協(xié)議。XA 中大致分為兩部分：事務(wù)管理器和本地資源管理器。

其中本地資源管理器往往由數(shù)據(jù)庫實(shí)現(xiàn)，比如 Oracle、DB2 這些商業(yè)數(shù)據(jù)庫都實(shí)現(xiàn)了 XA 接口，而事務(wù)管理器作為全局的調(diào)度者，負(fù)責(zé)各個本地資源的提交和回滾。

XA 協(xié)議通常包含兩階段提交(2PC)和三階段提交(3PC)兩種實(shí)現(xiàn)。兩階段提交顧名思義就是要進(jìn)行兩個階段的提交：第一階段，準(zhǔn)備階段(投票階段);第二階段，提交階段(執(zhí)行階段)。

實(shí)現(xiàn)過程如下所示：

二階段提交看似能夠提供原子性的操作，但它存在著一些缺陷，三段提交(3PC)是對兩段提交(2PC)的一種升級優(yōu)化，有興趣的可以深入了解一下，這里不再贅述。

TCC(Try-Confirm-Cancel)：是 Try、Commit、Cancel 三種指令的縮寫，又被稱補(bǔ)償事務(wù)。

其邏輯模式類似于 XA 兩階段提交，事務(wù)處理流程也很相似，但 2PC 是應(yīng)用于在 DB 層面，TCC 則可以理解為在應(yīng)用層面的 2PC，是需要我們編寫業(yè)務(wù)邏輯來實(shí)現(xiàn)。

TCC 它的核心思想是：“針對每個操作都要注冊一個與其對應(yīng)的確認(rèn)(Try)和補(bǔ)償(Cancel)”。

消息事務(wù)：所謂的消息事務(wù)就是基于消息隊(duì)列的兩階段提交，本質(zhì)上是對消息隊(duì)列的一種特殊利用。

它是將本地事務(wù)和發(fā)消息放在了一個分布式事務(wù)里，保證要么本地操作成功成功并且對外發(fā)消息成功，要么兩者都失敗。

基于消息隊(duì)列的兩階段提交往往用在高并發(fā)場景下，將一個分布式事務(wù)拆成一個消息事務(wù)(A 系統(tǒng)的本地操作+發(fā)消息)+B 系統(tǒng)的本地操作。

其中 B 系統(tǒng)的操作由消息驅(qū)動，只要消息事務(wù)成功，那么 A 操作一定成功，消息也一定發(fā)出來了。

這時候 B 會收到消息去執(zhí)行本地操作，如果本地操作失敗，消息會重投，直到 B 操作成功，這樣就變相地實(shí)現(xiàn)了 A 與 B 的分布式事務(wù)。

原理如下：

雖然上面的方案能夠完成 A 和 B 的操作，但是 A 和 B 并不是強(qiáng)一致的，而是最終一致(Eventually consistent)的。而這也是滿足 BASE 理論的要求的。

這里引申一下，BASE 是 Basically Available(基本可用)、Soft state(軟狀態(tài))和 Eventually consistent(最終一致性)三個短語的縮寫。

BASE 理論是對 CAP 中 AP(CAP 已經(jīng)被證實(shí)一個分布式系統(tǒng)最多只能同時滿足 CAP 三項(xiàng)中的兩項(xiàng))的一個擴(kuò)展，通過犧牲強(qiáng)一致性來獲得可用性。

當(dāng)出現(xiàn)故障允許部分不可用但要保證核心功能可用，允許數(shù)據(jù)在一段時間內(nèi)是不一致的，但最終達(dá)到一致狀態(tài)。滿足 BASE 理論的事務(wù)，我們稱之為“柔性事務(wù)”。

②什么是 Exactly-once (精確一次)語義?

在分布式系統(tǒng)中，任何節(jié)點(diǎn)都有可能出現(xiàn)異常甚至宕機(jī)。在消息隊(duì)列中也一樣，當(dāng) Producer 在生產(chǎn)消息時，可能會發(fā)生 Broker 宕機(jī)不可用，或者網(wǎng)絡(luò)突然中斷等異常情況。

根據(jù)在發(fā)生異常時 Producer 處理消息的方式，系統(tǒng)可以具備以下三種消息語義。

At-least-once(至少一次)語義：Producer 通過接收 Broker 的 ACK(消息確認(rèn))通知來確保消息成功寫入 Topic。

然而，當(dāng) Producer 接收 ACK 通知超時，或者收到 Broker 出錯信息時，會嘗試重新發(fā)送消息。

如果 Broker 正好在成功把消息寫入到 Topic，但還沒有給 Producer 發(fā)送 ACK 時宕機(jī)，Producer 重新發(fā)送的消息會被再次寫入到 Topic，最終導(dǎo)致消息被重復(fù)分發(fā)至 Consumer。即：消息不會丟失，但有可能被重復(fù)發(fā)送。

At-most-once(最多一次)語義：當(dāng) Producer 在接收 ACK 超時，或者收到 Broker 出錯信息時不重發(fā)消息，那就有可能導(dǎo)致這條消息丟失，沒有寫入到 Topic 中，也不會被 Consumer 消費(fèi)到。

在某些場景下，為了避免發(fā)生重復(fù)消費(fèi)，我們可以容許消息丟失的發(fā)生。即：消息可能會丟失，但絕不會被重復(fù)發(fā)送。

Exactly-once(精確一次)語義：Exactly-once 語義保證了即使 Producer 多次發(fā)送同一條消息到服務(wù)端，服務(wù)端也僅僅會記錄一次。

Exactly-once 語義是最可靠的，同時也是最難理解的。Exactly-once 語義需要消息隊(duì)列服務(wù)端，消息生產(chǎn)端和消費(fèi)端應(yīng)用三者的協(xié)同才能實(shí)現(xiàn)。

比如，當(dāng)消費(fèi)端應(yīng)用成功消費(fèi)并且 ACK 了一條消息之后，又把消費(fèi)位點(diǎn)回滾到之前的一個消息 ID，那么從那個消息 ID 往后的所有消息都會被消費(fèi)端應(yīng)用重新消費(fèi)到。即：消息不會丟失，也不會被重復(fù)發(fā)送。

RocketMQ、Kafka、Pulsar 事務(wù)消息

①RocketMQ 的事務(wù)消息

RocketMQ 在 4.3.0 版中已經(jīng)支持分布式事務(wù)消息，這里 RocketMQ 采用了 2PC 的思想來實(shí)現(xiàn)了提交事務(wù)消息，同時增加一個補(bǔ)償邏輯來處理二階段超時或者失敗的消息。

流程如下圖所示：

其具體工作流程分為正常事務(wù)消息的發(fā)送及提交和不正常情況下事務(wù)消息的補(bǔ)償流程：

• 在消息隊(duì)列上開啟一個事務(wù)主題。
• 事務(wù)中第一個執(zhí)行的服務(wù)發(fā)送一條“半消息”(半消息和普通消息的唯一區(qū)別是，在事務(wù)提交之前，對于消費(fèi)者來說，這個消息是不可見的)給消息隊(duì)列。
• 半消息發(fā)送成功后，發(fā)送半消息的服務(wù)就會開始執(zhí)行本地事務(wù)，根據(jù)本地事務(wù)執(zhí)行結(jié)果來決定事務(wù)消息提交或者回滾。
• 本地事務(wù)成功后會讓這個“半消息”變成正常消息，供分布式事務(wù)后面的步驟執(zhí)行自己的本地事務(wù)。

這里的事務(wù)消息，Producer 不會因?yàn)?Consumer 消費(fèi)失敗而做回滾，采用事務(wù)消息的應(yīng)用，其所追求的是高可用和最終一致性，消息消費(fèi)失敗的話，RocketMQ 自己會負(fù)責(zé)重推消息，直到消費(fèi)成功。

補(bǔ)償流程：RocketMQ 提供事務(wù)反查來解決異常情況，如果 RocketMQ 沒有收到提交或者回滾的請求，Broker 會定時到生產(chǎn)者上去反查本地事務(wù)的狀態(tài)，然后根據(jù)生產(chǎn)者本地事務(wù)的狀態(tài)來處理這個“半消息”是提交還是回滾。

值得注意的是我們需要根據(jù)自己的業(yè)務(wù)邏輯來實(shí)現(xiàn)反查邏輯接口，然后根據(jù)返回值 Broker 決定是提交還是回滾。

而且這個反查接口需要是無狀態(tài)的，請求到任意一個生產(chǎn)者節(jié)點(diǎn)都會返回正確的數(shù)據(jù)。

其中，補(bǔ)償流程用于解決消息 Commit 或者 Rollback 發(fā)生超時或者失敗的情況。在 RocketMQ 事務(wù)消息的主要流程中，一階段的消息如何對用戶不可見。

其中，事務(wù)消息相對普通消息最大的特點(diǎn)就是一階段發(fā)送的消息對用戶是不可見的。

那么，如何做到寫入消息但是對用戶不可見呢?RocketMQ 事務(wù)消息的做法是：如果消息是“半消息”，將備份原消息的主題與消息消費(fèi)隊(duì)列，然后改變主題為 RMQ_SYS_TRANS_HALF_TOPIC。

由于消費(fèi)組未訂閱該主題，故消費(fèi)端無法消費(fèi)“半消息”的消息，然后 RocketMQ 會開啟一個定時任務(wù)，從 Topic 為 RMQ_SYS_TRANS_HALF_TOPIC 中拉取消息進(jìn)行消費(fèi)。

根據(jù)生產(chǎn)者組獲取一個服務(wù)提供者發(fā)送回查事務(wù)狀態(tài)請求，根據(jù)事務(wù)狀態(tài)來決定是提交或回滾消息。

講到這里大家就明白了，這里說的就是上文提到分布式事務(wù)中的消息事務(wù)，目的是在分布式事務(wù)中實(shí)現(xiàn)系統(tǒng)的最終一致性。

②Kafka 的事務(wù)消息

與 RocketMQ 的事務(wù)消息用途不同，Kafka 的事務(wù)基本上是配合其冪等機(jī)制來實(shí)現(xiàn) Exactly-once(見上文)語義的。

開發(fā)此功能的原因可以總結(jié)如下：

流處理的需求：隨著流處理的興起，對具有更強(qiáng)處理保證的流處理應(yīng)用的需求也在增長。

例如，在金融行業(yè)，金融機(jī)構(gòu)使用流處理引擎為用戶處理借款和信貸。這種類型的用例要求每條消息都只處理一次，無一例外。

換句話說，如果流處理應(yīng)用程序消費(fèi)消息 A 并將結(jié)果作為消息 B(B = f(A))，那么恰好一次處理保證意味著當(dāng)且僅當(dāng) B 被成功生產(chǎn)后 A 才能被標(biāo)記為消費(fèi)，反之亦然。

事務(wù) API 使流處理應(yīng)用程序能夠在一個原子操作中使用、處理和生成消息。這意味著，事務(wù)中的一批消息可以從許多主題分區(qū)接收、生成和確認(rèn)。一個事務(wù)涉及的所有操作都作為整體成功或失敗。

目前，Kafka 默認(rèn)提供的交付可靠性保障是 At-least-once。如果消息成功“提交”，但 Broker 的應(yīng)答沒有成功發(fā)送回 Producer 端(比如網(wǎng)絡(luò)出現(xiàn)瞬時抖動)，那么 Producer 就無法確定消息是否真的提交成功了。

因此，它只能選擇重試，這就是 Kafka 默認(rèn)提供 At-least-once 保障的原因，不過這會導(dǎo)致消息重復(fù)發(fā)送。

大部分用戶還是希望消息只會被交付一次，這樣的話，消息既不會丟失，也不會被重復(fù)處理。

或者說，即使 Producer 端重復(fù)發(fā)送了相同的消息，Broker 端也能做到自動去重。

在下游 Consumer 看來，消息依然只有一條。那么問題來了，Kafka 是怎么做到精確一次的呢?

簡單來說，這是通過兩種機(jī)制：

• 冪等性(Idempotence)
• 事務(wù)(Transaction)

冪等性 Producer：“冪等”這個詞原是數(shù)學(xué)領(lǐng)域中的概念，指的是某些操作或函數(shù)能夠被執(zhí)行多次，但每次得到的結(jié)果都是不變的。

冪等性有很多好處，其最大的優(yōu)勢在于我們可以安全地重試任何冪等性操作，反正它們也不會破壞我們的系統(tǒng)狀態(tài)。

如果是非冪等性操作，我們還需要擔(dān)心某些操作執(zhí)行多次對狀態(tài)的影響，但對于冪等性操作而言，我們根本無需擔(dān)心此事。

在 Kafka 中，Producer 默認(rèn)不是冪等性的，但我們可以創(chuàng)建冪等性 Producer。它其實(shí)是 0.11.0.0 版本引入的新功能。

enable.idempotence 被設(shè)置成 true 后，Producer 自動升級成冪等性 Producer，其他所有的代碼邏輯都不需要改變。

Kafka 自動幫你做消息的重復(fù)去重。Kafka 為了實(shí)現(xiàn)冪等性，它在底層設(shè)計架構(gòu)中引入了 ProducerID 和 SequenceNumber。

ProducerID：在每個新的 Producer 初始化時，會被分配一個唯一的 ProducerID，用來標(biāo)識本次會話。

SequenceNumber：對于每個 ProducerID，Producer 發(fā)送數(shù)據(jù)的每個 Topic 和 Partition 都對應(yīng)一個從 0 開始單調(diào)遞增的 SequenceNumber 值。

Broker 在內(nèi)存維護(hù)(pid，seq)映射，收到消息后檢查 seq。Producer 在收到明確的的消息丟失 ack，或者超時后未收到 ack，要進(jìn)行重試。

• new_seq=old_seq+1：正常消息。
• new_seq<=old_seq：重復(fù)消息。
• new_seq>old_seq+1：消息丟失。

另外我們需要了解冪等性 Producer 的作用范圍。首先，它只能保證單分區(qū)上的冪等性，即一個冪等性 Producer 能夠保證某個主題的一個分區(qū)上不出現(xiàn)重復(fù)消息，它無法實(shí)現(xiàn)多個分區(qū)的冪等性。

其次，它只能實(shí)現(xiàn)單會話上的冪等性，不能實(shí)現(xiàn)跨會話的冪等性。這里的會話，你可以理解為 Producer 進(jìn)程的一次運(yùn)行。當(dāng)你重啟了 Producer 進(jìn)程之后，這種冪等性保證就喪失了。

如果想實(shí)現(xiàn)多分區(qū)以及多會話上的消息無重復(fù)，應(yīng)該怎么做呢?答案就是事務(wù)(transaction)或者依賴事務(wù)型 Producer。這也是冪等性 Producer 和事務(wù)型 Producer 的最大區(qū)別。

事務(wù)型 Producer：能夠保證將消息原子性地寫入到多個分區(qū)中。這批消息要么全部寫入成功，要么全部失敗。

另外，事務(wù)型 Producer 也不受進(jìn)程的重啟影響。Producer 重啟后，Kafka 依然保證它們發(fā)送消息的 Exactly-once 處理。

和普通 Producer 代碼相比，事務(wù)型 Producer 的顯著特點(diǎn)是調(diào)用了一些事務(wù) API。

如 initTransaction、beginTransaction、commitTransaction 和 abortTransaction，它們分別對應(yīng)事務(wù)的初始化、事務(wù)開始、事務(wù)提交以及事務(wù)終止。

Kafka 事務(wù)消息是由 Producer、事務(wù)協(xié)調(diào)器、Broker、組協(xié)調(diào)器、Consumer 等共同參與實(shí)現(xiàn)的。

Producer：為 Producer 指定固定的 TransactionalId(事務(wù) id)，可以穿越 Producer 的多次會(Producer 重啟/斷線重連)中，持續(xù)標(biāo)識 Producer 的身份。

每個生產(chǎn)者增加一個 epoch。用于標(biāo)識同一個 TransactionalId 在一次事務(wù)中的 epoch，每次初始化事務(wù)時會遞增，從而讓服務(wù)端可以知道生產(chǎn)者請求是否舊的請求。

使用 epoch 標(biāo)識 Producer 的每一次“重生”，可以防止同一 Producer 存在多個會話。

Producer 遵從冪等消息的行為，并在發(fā)送的 BatchRecord 中增加事務(wù) id 和 epoch。

事務(wù)協(xié)調(diào)器(Transaction Coordinator)：引入事務(wù)協(xié)調(diào)器，類似于消費(fèi)組負(fù)載均衡的協(xié)調(diào)者，每一個實(shí)現(xiàn)事務(wù)的生產(chǎn)端都被分配到一個事務(wù)協(xié)調(diào)者。以兩階段提交的方式，實(shí)現(xiàn)消息的事務(wù)提交。

事務(wù)協(xié)調(diào)器使用一個特殊的 Topic：即事務(wù) Topic，事務(wù) Topic 本身也是持久化的，日志信息記錄事務(wù)狀態(tài)信息，由事務(wù)協(xié)調(diào)者寫入。

事務(wù)協(xié)調(diào)器通過 RPC 調(diào)用，協(xié)調(diào) Broker 和 Consumer 實(shí)現(xiàn)事務(wù)的兩階段提交。

每一個 Broker 都會啟動一個事務(wù)協(xié)調(diào)器，使用 hash(TransactionalId)確定 Producer 對應(yīng)的事務(wù)協(xié)調(diào)器，使得整個集群的負(fù)載均衡。

Broker：引入控制消息(Control Messages)：這些消息是客戶端產(chǎn)生的并寫入到主題的特殊消息，但對于使用者來說不可見。它們是用來讓 Broker 告知消費(fèi)者之前拉取的消息是否被原子性提交。

Broker 處理事務(wù)協(xié)調(diào)器的 commit/abort 控制消息，把控制消息向正常消息一樣寫入 Topic(圖中標(biāo) c 的消息，和正常消息交織在一起，用來確認(rèn)事務(wù)提交的日志偏移)，并向前推進(jìn)消息提交偏移 hw。

組協(xié)調(diào)器：如果在事務(wù)過程中，提交了消費(fèi)偏移，組協(xié)調(diào)器在 offset log 中寫入事務(wù)消費(fèi)偏移。當(dāng)事務(wù)提交時，在 offset log 中寫入事務(wù) offset 確認(rèn)消息。

Consumer：Consumer 過濾未提交消息和事務(wù)控制消息，使這些消息對用戶不可見。

有兩種實(shí)現(xiàn)方式：

Consumer 緩存方式：設(shè)置 isolation.level=read_uncommitted，此時 topic 的所有消息對 Consumer 都可見。

Consumer 緩存這些消息，直到收到事務(wù)控制消息。若事務(wù) commit，則對外發(fā)布這些消息;若事務(wù) abort，則丟棄這些消息。

Broker 過濾方式：設(shè)置 isolation.level=read_committed，此時 topic 中未提交的消息對 Consumer 不可見，只有在事務(wù)結(jié)束后，消息才對 Consumer 可見。

Broker 給 Consumer 的 BatchRecord 消息中，會包含以列表，指明哪些是“abort”事務(wù)，Consumer 丟棄 abort 事務(wù)的消息即可。

因?yàn)槭聞?wù)機(jī)制會影響消費(fèi)者所能看到的消息的范圍，它不只是簡單依賴高水位來判斷。

它依靠一個名為 LSO(Log Stable Offset)的位移值來判斷事務(wù)型消費(fèi)者的可見性。

③Pulsar 的事務(wù)消息

Apache Pulsar 在 2.8.0 正式支持了事務(wù)相關(guān)的功能，Pulsar 這里提供的事務(wù)區(qū)別于 RocketMQ 中 2PC 那種事務(wù)的實(shí)現(xiàn)方式，沒有本地事務(wù)回查的機(jī)制，更類似于 Kafka 的事務(wù)實(shí)現(xiàn)機(jī)制。

Apache Pulsar 中的事務(wù)主要用來保證類似 Pulsar Functions 這種流計算場景中 Exactly-once 語義的實(shí)現(xiàn)。

這也符合 Apache Pulsar 本身 Event Streaming 的定位，即保證端到端(End-to-End)的事務(wù)實(shí)現(xiàn)的語義。

在 Pulsar 中，對于事務(wù)語義是這樣定義的：允許事件流應(yīng)用將消費(fèi)、處理、生產(chǎn)消息整個過程定義為一個原子操作，即生產(chǎn)者或消費(fèi)者能夠處理跨多個主題和分區(qū)的消息，并確保這些消息作為一個單元被處理。

Pulsar 事務(wù)具有以下語義：

• 事務(wù)中的所有操作都作為一個單元提交。要么提交所有消息，要么都不提交。
• 每條消息只寫入或處理一次，不會丟失數(shù)據(jù)或重復(fù)(即使發(fā)生故障)。
• 如果事務(wù)中止，則此事務(wù)中的所有寫入和確認(rèn)都將回滾。

事務(wù)中的批量消息可以被以多分區(qū)接收、生產(chǎn)和確認(rèn)：

• 消費(fèi)者只能讀取已提交(確認(rèn))的消息。換句話說，Broker 不傳遞屬于打開事務(wù)的事務(wù)消息或?qū)儆谥兄故聞?wù)的消息。
• 跨多個分區(qū)的消息寫入是原子性的。
• 跨多個訂閱的消息確認(rèn)是原子性的。訂閱下的消費(fèi)者在確認(rèn)帶有事務(wù) ID 的消息時，只會成功確認(rèn)一次消息。

Pulsar 事務(wù)消息由以下幾個關(guān)鍵點(diǎn)構(gòu)成：

事務(wù) ID(TxnID)：標(biāo)識 Pulsar 中的唯一事務(wù)。事務(wù) ID 長度是 128-bit。最高 16 位保留給事務(wù)協(xié)調(diào)器的 ID，其余位用于每個事務(wù)協(xié)調(diào)器中單調(diào)遞增的數(shù)字。

事務(wù)協(xié)調(diào)器(TC)：是運(yùn)行在 Pulsar Broker 中的一個模塊。它維護(hù)事務(wù)的整個生命周期，并防止事務(wù)進(jìn)入錯誤狀態(tài);它處理事務(wù)超時，并確保事務(wù)在事務(wù)超時后中止。

事務(wù)日志：所有事務(wù)元數(shù)據(jù)都保存在事務(wù)日志中。事務(wù)日志由 Pulsar 主題記錄。如果事務(wù)協(xié)調(diào)器崩潰，它可以從事務(wù)日志恢復(fù)事務(wù)元數(shù)據(jù)。

事務(wù)日志存儲事務(wù)狀態(tài)，而不是事務(wù)中的實(shí)際消息(實(shí)際消息存儲在實(shí)際的主題分區(qū)中)。

事務(wù)緩存：向事務(wù)內(nèi)的主題分區(qū)生成的消息存儲在該主題分區(qū)的事務(wù)緩沖區(qū)(TB)中。

在提交事務(wù)之前，事務(wù)緩沖區(qū)中的消息對消費(fèi)者不可見。當(dāng)事務(wù)中止時，事務(wù)緩沖區(qū)中的消息將被丟棄。

事務(wù)緩沖區(qū)將所有正在進(jìn)行和中止的事務(wù)存儲在內(nèi)存中。所有消息都發(fā)送到實(shí)際的分區(qū) Pulsar 主題。

提交事務(wù)后，事務(wù)緩沖區(qū)中的消息對消費(fèi)者具體化(可見)。事務(wù)中止時，事務(wù)緩沖區(qū)中的消息將被丟棄。

待確認(rèn)狀態(tài)：掛起確認(rèn)狀態(tài)在事務(wù)完成之前維護(hù)事務(wù)中的消息確認(rèn)。如果消息處于掛起確認(rèn)狀態(tài)，則在該消息從掛起確認(rèn)狀態(tài)中移除之前，其他事務(wù)無法確認(rèn)該消息。

掛起的確認(rèn)狀態(tài)被保留到掛起的確認(rèn)日志中(cursor ledger)。新啟動的 broker 可以從掛起的確認(rèn)日志中恢復(fù)狀態(tài)，以確保狀態(tài)確認(rèn)不會丟失。

處理流程一般分為以下幾個步驟：

• 開啟事務(wù)。
• 使用事務(wù)發(fā)布消息。
• 使用事務(wù)確認(rèn)消息。
• 結(jié)束事務(wù)。

Pulsar 的事務(wù)處理流程與 Kafka 的事務(wù)處理思路大致上保持一致，大家都有一個 TC 以及對應(yīng)的一個用于持久化 TC 所有操作的 Topic 來記錄所有事務(wù)狀態(tài)變更的請求。

同樣的在事務(wù)開始階段也都有一個專門的 Topic 來去查詢 TC 對應(yīng)的 Owner Broker 的位置在哪里。

不同的是：

• Kafka 中對于未確認(rèn)的消息是維護(hù)在 Broker 端的，但是 Pulsar 的是維護(hù)在 Client 端的，通過 Transaction Timeout 來決定這個事務(wù)是否執(zhí)行成功，所以有了 Transaction Timeout 的存在之后，就可以確保 Client 和 Broker 側(cè)事務(wù)處理的一致性。
• 由于 Kafka 本身沒有單條消息的 Ack，所以 Kafka 的事務(wù)處理只能是順序執(zhí)行的，當(dāng)一個事務(wù)請求被阻塞之后，會阻塞后續(xù)所有的事務(wù)請求，但是 Pulsar 是可以對消息進(jìn)行單條 Ack 的，所以在這里每一個事務(wù)的 Ack 動作是獨(dú)立的，不會出現(xiàn)事務(wù)阻塞的情況。

結(jié)論

RocketMQ 和 Kafka/Pulsar 的事務(wù)消息實(shí)用的場景是不一樣的。

RocketMQ 中的事務(wù)，它解決的問題是，確保執(zhí)行本地事務(wù)和發(fā)消息這兩個操作，要么都成功，要么都失敗。

并且 RocketMQ 增加了一個事務(wù)反查的機(jī)制，來盡量提高事務(wù)執(zhí)行的成功率和數(shù)據(jù)一致性。

Kafka 中的事務(wù)，它解決的問題是，確保在一個事務(wù)中發(fā)送的多條消息，要么都成功，要么都失敗。

這里面的多條消息不一定要在同一個主題和分區(qū)中，可以是發(fā)往多個主題和分區(qū)的消息。

當(dāng)然也可以在 Kafka 事務(wù)執(zhí)行過程中開啟本地事務(wù)來實(shí)現(xiàn)類似 RocketMQ 事務(wù)消息的效果。

但是 Kafka 是沒有事務(wù)消息反查機(jī)制的，它是直接拋出異常的，用戶可以根據(jù)異常來實(shí)現(xiàn)自己的重試等方法保證事務(wù)正常運(yùn)行。

它們的共同點(diǎn)就是：都是通過兩階段提交來實(shí)現(xiàn)事務(wù)的，事務(wù)消息都保存在單獨(dú)的主題上。

不同的地方就是 RocketMQ 是通過“半消息”來實(shí)現(xiàn)的，Kafka 是直接將消息發(fā)送給對應(yīng)的 topic，通過客戶端來過濾實(shí)現(xiàn)的。

而且它們兩個使用的場景區(qū)別是非常之大的，RockteMQ 主要解決的是基于本地事務(wù)和消息的數(shù)據(jù)一致性，而 Kafka 的事務(wù)則是用于實(shí)現(xiàn)它的 Exactly-once 機(jī)制，應(yīng)用于實(shí)時流計算的場景中。

Pulsar 的事務(wù)消息和 Kafka 應(yīng)用場景和語義類似，只是由于底層實(shí)現(xiàn)機(jī)制有差別，在一些細(xì)節(jié)上有區(qū)別。

相信看到這里就非常清楚了，對于事務(wù)消息如何選型和應(yīng)用，首先要明白你的業(yè)務(wù)需求是什么。

是要實(shí)現(xiàn)分布式事務(wù)的最終一致性，還是要實(shí)現(xiàn) Exactly-once (精確一次)語義?明白之后需求，選擇什么組件就十分明確了。

作者：劉若愚

簡介：微信支付后臺開發(fā)工程師，碩士畢業(yè)于北京大學(xué)。深度參與騰訊 WXG 境外支付團(tuán)隊(duì)多個重要業(yè)務(wù)的研發(fā)工作，有豐富的后臺開發(fā)經(jīng)驗(yàn)。騰訊技術(shù)分享達(dá)人，社會招聘伯樂。

編輯：陶家龍

出處：轉(zhuǎn)載自公眾號云加社區(qū)(ID：QcloudCommunity)

文章名稱：新同事上來就用Kafka，瑟瑟發(fā)抖...
網(wǎng)站URL：http://m.fisionsoft.com.cn/article/cccidse.html