我們即將開始一個新的部落格系列，重點介紹在 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 Framework 中提供的傳統交易支援來使用它。Spring Cloud Stream 中的 Kafka binder 進一步建立在 Spring for Apache Kafka 的這種支援之上，使得在 Spring Cloud Stream Kafka 應用程式中使用相同的支援成為可能。在本部落格系列的第一部分中，我們將簡要介紹 Kafka 交易、一些使用案例分析（在這些案例中，依賴交易會很有幫助），以及 Apache Kafka 和 Spring 生態系統中的交易建構區塊。

在許多使用案例中，在 Apache Kafka 中以交易方式發佈、消費和處理記錄變得必要。當在生產者啟動的應用程式或以交易方式實作消費-處理-生產模式的程序中以交易方式生產記錄時，它們會以原子方式寫入 Kafka。如果出現問題，整個程序將會回滾，並且交易不會提交。需要記住的一件事是，與支援交易的關聯式資料庫不同，在這種資料庫中，當發生此類交易回滾時，不會持久化任何記錄，而 Apache Kafka 仍然會將記錄發佈到主題分割區。這種行為是由於 Apache Kafka 的基本唯附加不可變日誌架構，該架構不允許任何記錄修改，例如在將記錄新增到記錄日誌後移除記錄。有人可能會想知道使用交易的好處是什麼，因為當交易中止時，記錄可能會發佈到主題分割區，這可能會導致消費者看到它們。然而，具有適當隔離等級的消費者永遠不會看到回滾的記錄，即使來自回滾交易的記錄在主題分割區中。因此，從端對端角度來看，整個程序保證是完全交易性的。

交易使用案例

交易通常會在 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 來追蹤交易的所有方面，例如初始化、正在進行的進度、提交等等。

以下清單總結了重要的交易相關生產者 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 binder 這樣依賴它的框架時，它們帶來的好處是允許應用程式主要關注業務邏輯，因為框架會處理我們在上面看到的低階樣板交易序列。使用 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 Framework 中 PlatformTransactionManager 的實作。您可以將此交易管理員與 Transactional 註解一起使用，或在使用 TransactionTemplate 的本機交易中使用。KafkaTransactionManager 使用生產者工廠來建立 Kafka 生產者，並提供 API 來開始、提交和回滾交易。

KafkaResourceHolder

Spring for Apache Kafka 還提供了一個 KafkaResourceHolder，它持有 Kafka 生產者資源。Spring for Apache Kafka 中的 KafkaTemplate 觸發給定生產者工廠的 KafkaResourceHolder 在目前執行緒上的綁定。在消費者啟動的交易的情況下，訊息監聽器容器會執行此綁定，而生產者工廠與 KafkaTransactionManager 使用的生產者工廠相同。這樣，交易將相同的交易生產者用於所有發佈需求。

除了上述組件外，Spring for Apache Kafka 還提供了其他用於處理交易相關問題的實用程式。當我們瀏覽本系列的以下章節時，我們會看到其中的一些。

在本部落格系列的第 2 部分中，我們將繼續討論在 Spring Cloud Stream 應用程式中使用交易的更實際的實作細節。