我們正在開始一個新的部落格系列，該系列專注於處理 Spring Cloud Stream Kafka 應用中的事務。本部落格系列涵蓋了使用 Spring Cloud Stream 和 Apache Kafka 編寫事務性應用程式的許多底層細節。希望透過本部落格系列，我們能為您提供足夠的資訊，以便為各種業務用例編寫事務性 Spring Cloud Stream Kafka 應用程式。

基本構建塊

Spring Cloud Stream Kafka 應用程式中的事務基礎支援主要來自 Apache Kafka 本身和 Spring for Apache Kafka 庫。然而，本系列部落格主要關注的是將此支援專門用於 Spring Cloud Stream。如果您熟悉 Apache Kafka 中的事務工作原理以及 Spring for Apache Kafka 如何使其在 Spring 中易於使用，那麼本系列會讓您感到賓至如歸。

雖然 Apache Kafka 提供了基礎的事務支援，但 Spring for Apache Kafka (又稱 Spring Kafka) 庫在 Spring 端擴充套件了此支援，透過利用 Spring 框架中傳統的事務支援，使其對 Spring 開發者來說更加自然。Spring Cloud Stream 中的 Kafka 繫結器在此基礎上進一步構建，使得在 Spring Cloud Stream Kafka 應用程式中可以使用相同的支援。在本系列部落格的第一部分，我們將簡要介紹 Kafka 事務，分析一些需要依賴事務的用例，以及 Apache Kafka 和 Spring 生態系統中的事務構建塊。

在 Apache Kafka 中，存在許多需要以事務方式釋出、消費和處理記錄的用例。當在生產者發起的應用程式或以事務方式實現“消費-處理-釋出”模式的程序中生成事務性記錄時，這些記錄會被原子地寫入 Kafka。如果出現任何問題，整個過程將被回滾，並且事務不會被提交。需要注意的一點是，與支援事務的關係型資料庫不同，在關係型資料庫中，如果發生事務回滾，則不會持久化任何記錄；而在 Apache Kafka 中，記錄仍會被髮布到主題分割槽。這是因為 Apache Kafka fundamentally 是一個基於日誌的、不可修改的、僅追加的架構，它不允許修改記錄，例如在將記錄新增到記錄日誌後將其移除。有人可能會想，既然在事務被中止時記錄可能被髮布到主題分割槽，從而可能導致消費者看到它們，那麼使用事務有什麼好處呢？然而，具有適當的隔離級別的消費者即使在回滾事務的記錄位於主題分割槽中時，也不會看到已回滾的記錄。因此，從端到端的角度來看，整個過程被保證是完全事務性的。

事務性用例

事務通常會增加 Kafka 應用程式的顯著開銷。在使用 Apache Kafka 的事務時，每條記錄都必須新增特殊的事務日誌，並將事務標記傳送到特殊的主題狀態主題等。所有這些步驟都需要時間和空間，增加了整體延遲。因此，每個應用程式都必須透過分析用例來仔細權衡對事務支援的需求。

事務主要提供了一種保護資料的方法，以提供ACID 功能。它透過提供原子性、一致性、資料隔離和永續性來確保資料完整性。

當今企業中有許多關鍵任務用例，在這些用例中，使用事務並依賴其帶來的 ACID 語義是高度可取的。關於何時使用事務並證明其帶來的開銷是合理的，沒有簡單明瞭的答案。您必須檢視應用程式並評估其風險。事務的典型經典示例是任何需要處理金融資料的場景。Bob 向 Alice 傳送錢，這一操作會從 Bob 的賬戶中扣款，Alice 的賬戶會得到記賬。如果在此過程中出現任何問題，整個過程將被回滾，彷彿什麼都沒有發生，因為我們不希望流程處於混亂狀態。如果該過程從 Bob 的賬戶扣款，但 Alice 未收到記賬（或反之），那將是一個問題。從 Apache Kafka 的角度來看，這裡有幾件事正在發生。首先，訊息來到 Kafka 處理器，用於從 Bob 的賬戶扣款以及接收方資訊。處理器處理資訊，然後傳送一條訊息到另一個主題，表明 Bob 的賬戶已發生扣款。在此之後，另一條訊息表明 Alice 已收到記賬。此過程中的各種操作需要複雜的協調，以確保一切按預期進行。每當我們有多個相關事件時，事務可能有助於確保資料完整性並提供 ACID 語義。在此示例中，單個事件本身意義不大，但它們共同構成了整個流程，並需要事務性來確保資料完整性。

如果我們想概括這種模式，我們可以說，任何時候我們有一個“消費-處理-釋出”模式，它是關鍵任務的，其中如果一個元件失敗，整個處理器都需要表現得好像它從未發生過一樣，那麼使用事務是一個值得考慮的潛在解決方案。

來自其他領域的更多高階示例

• 想象一個航空訂票系統，它需要釋出包含多個航段的預訂資訊。如果由於任何原因，系統無法釋出整個預訂，則需要中止該過程並重新開始。
• 一家經紀公司向結算中心傳送包含多個買入訂單的請求。假設該程序無法將單個訂單作為單一原子單元釋出到結算中心正在消費的訊息系統中。在這種情況下，經紀公司必須重新發送訂單。
• 一家傳送患者測試資料的醫療計費系統到保險公司，必須將患者的各種相關測試釋出到訊息系統中。
• 一個線上遊戲系統需要跟蹤玩家在遊戲中的位置，並將它們事務性地傳送到一箇中央伺服器，以確保所有玩家看到正確的座標，而不是部分更新的位置。
• 零售商的庫存補貨系統需要將各種相關產品狀態的資訊作為單一原子單元傳送。
• 一個線上電子商務訂單系統，在一次原子聚合操作中釋出訂單詳情（如訂單條目、賬戶持有人資訊、運輸資訊等）。

與外部資料庫同步

另一個事務變得很有用的用例是，當您必須與其他事務性系統同步時。除了釋出到 Kafka 之外，假設您還必須在關係型資料庫中持久化記錄或某些派生資訊，所有這些都在一次原子操作中完成。如果一個系統未能傳送資料，我們必須回滾。如果您每次只發布一條記錄到 Kafka，並且沒有其他操作，也沒有其他相關操作，那麼您就不需要使用事務，這一點我們將在本系列部落格的下一部分中看到。然而，即使您只發布到 Kafka 主題一次，但將關係型資料庫操作作為同一過程的一部分，使用事務就變得必要，以確保資料完整性。

釋出到多個 Kafka 主題

在僅生產者應用程式中，事務的另一個用例是釋出到多個 Kafka 主題。假設您有一些關鍵的業務資料，形式為重要通知（例如訂單詳情），您希望將其釋出到多個 Kafka 主題，訂單詳情的一部分發布到訂單主題，另一部分發布到運輸主題。在這種情況下，我們可以使用事務來確保端到端的資料完整性。

對上述事務性用例的概括

上述用例列表並非詳盡無遺，其中事務是必需的。在當今的企業中，存在許多其他用例，它們與我們所考察的用例的總體方向並無太大差異，這些用例需要訊息系統的事務性處理。

以下列表總結了 Apache Kafka 中事務可能有所幫助的通用用例：

• “消費-處理-釋出”系統，需要將記錄作為單一原子單元釋出，並提供精確一次語義的交付保證。
• 多個相關的釋出事件，單獨來看沒有意義。
• 將資料作為單一原子單元釋出到多個主題。
• 與外部事務管理器同步。

這裡是所有這些不同情況的圖示表示。它涵蓋了我們上面考慮的場景，例如“消費-處理-釋出”、“多個生產者”、“與外部事務同步”等。處理器從入站主題消費資料，執行業務邏輯，將某些資訊持久化到資料庫系統，然後釋出到多個 Kafka 主題。

Apache Kafka 中的事務

有大量的文獻可以研究 Apache Kafka 中事務工作原理的底層細節，本文可以為您介紹這些細節。但是，從非常高的層面簡要了解實現事務性的 Kafka 客戶端 API 仍然是值得的。需要注意的一點是，對於普通消費者來說，Kafka 中沒有所謂的事務性消費者，但有感知事務的消費者。消費者透過設定隔離級別來實現這種事務感知。預設情況下，Kafka 中的消費者可以看到所有記錄，包括上游生產者未提交的記錄，因為 Kafka 消費者中的預設隔離級別是 **read_uncommitted**。Kafka 消費者必須使用 **read_committed** 隔離級別才能提供端到端的事務語義。我們將在本系列部落格的後續章節中瞭解如何在 Spring Cloud Stream 中實現這一點。

在生產者方面，應用程式可以依賴 Kafka 客戶端提供的幾個 API 方法。讓我們來看一下重要的幾個。

為了使應用程式具有事務性，Kafka 客戶端需要一個事務 ID。應用程式透過一個名為 **transactional.id** 的 Kafka 生產者屬性來提供此 ID，事務協調器使用此 ID 來透過註冊它來啟動事務。事務協調器使用此 ID 來跟蹤事務的所有方面，例如初始化、進行中、提交等。

以下列表總結了關鍵的事務相關生產者 API 方法。

Producer#initTransactions() - 對每個生產者呼叫一次以啟動事務支援。初始化 Kafka 事務。

Producer#beginTransaction() - 在傳送記錄之前開始事務。

Producer#sendOffsetsToTransaction() - 將已消費記錄的偏移量傳送到事務。

Producer#commitTransaction() - 提交事務。

Producer#abortTransaction() - 中止事務。

在傳送記錄之前，我們需要初始化並開始事務。然後，它繼續進行資料處理。如果我們消費了一條記錄來執行此釋出，我們必須使用生產者將已消費記錄的偏移量傳送到事務。在此之後，事務提交或中止操作可以繼續（commitTransaction 或 abortTransaction）。當我們呼叫 commitTransaction 方法時，Kafka 客戶端就會將偏移量原子地傳送到 consumer_offsets 主題。

Spring for Apache Kafka 中的事務支援

當使用 Spring for Apache Kafka 或依賴於它的 Spring Cloud Stream Kafka 繫結器等框架時，它們帶來了讓應用程式主要關注業務邏輯的好處，因為框架處理了我們上面看到的低階樣板事務序列。使用 Spring for Apache Kafka 或另一個框架（如使用它的 Spring Cloud Stream）是有益的，因為它使我們不必擔心編寫低階樣板序列（如上所述）來確保所有事務步驟都成功。正如您可以想象的那樣，這裡有很多移動的部分，如果您遺漏了一個步驟或沒有按照預期執行一個步驟，它可能會導致應用程式容易出錯。在 Spring 的情況下，我們提到的框架代表應用程式開發人員處理這些問題。讓我們簡要看看它是如何做到的。

Spring for Apache Kafka 框架透過提供一種熟悉 Spring 開發者的統一事務程式設計模型來隱藏所有這些底層細節。結果是，當使用 Spring for Apache Kafka 或其他框架（如 Spring Cloud Stream）時，應用程式可以簡單地專注於應用程式的業務邏輯，而不是處理複雜的低階事務相關問題。

KafkaTransactionManager

Spring for Apache Kafka 是如何提供這種統一事務程式設計模型的？簡而言之，Spring 開發者傳統上使用 @Transactional 註解或程式設計方法，例如直接在應用程式中使用 TransactionTemplate 來建立本地事務。這些機制需要一個事務管理器實現來驅動事務性方面。Spring for Apache Kafka 提供了一個事務管理器實現。**KafkaTransactionManager** 是 Spring 框架中 **PlatformTransactionManager** 的一個實現。您可以將此事務管理器與 @Transactional 註解一起使用，或者透過使用 TransactionTemplate 在本地事務中使用。KafkaTransactionManager 使用生產者工廠來建立 Kafka 生產者，並提供用於開始、提交和回滾事務的 API。

KafkaResourceHolder

Spring for Apache Kafka 還提供了一個 **KafkaResourceHolder**，它持有 Kafka 生產者資源。Spring for Apache Kafka 中的 KafkaTemplate 會觸發一個 KafkaResourceHolder 在當前執行緒上繫結到一個給定的生產者工廠。在消費者發起的事務中，訊息監聽器容器執行此繫結，並且生產者工廠與 KafkaTransactionManager 使用的工廠相同。這樣，事務就可以為所有釋出需求使用同一個事務性生產者。

除了上述元件之外，Spring for Apache Kafka 還提供了其他用於處理事務相關問題的實用工具。隨著本系列後續章節的深入，我們會根據需要介紹其中的一些。

在本系列部落格的第二部分，我們將繼續探討 Spring Cloud Stream 應用程式中事務使用的更實際的實現細節。