世界已經(jīng)邁進"移動"時代，現(xiàn)在應用程序必須能夠實時提供數(shù)據(jù)，這不僅包括數(shù)據(jù)庫表中存儲的重要最終結果，還包括用戶使用應用程序時執(zhí)行的所有操作。任何可用信息，例如，用戶點擊量、日志數(shù)據(jù)或傳感器數(shù)據(jù)都可用于改善用戶體驗、生成報告、向機器學習系統(tǒng)提供數(shù)據(jù)，等等。現(xiàn)如今，開發(fā)者必須關注基于實時事件流的系統(tǒng)。

下圖展示了基于事件流處理的架構示例。

Apache Kafka 已成為構建基于事件的高度可擴展系統(tǒng)的首選解決方案。Kafka 可為事件流平臺提供快速升級換代的功能，以供開發(fā)者在現(xiàn)代化業(yè)務解決方案中使用。然而，開發(fā)者通常需要將現(xiàn)有 Java EE 業(yè)務解決方案(基于 IBM MQ 或 IBM WebSphere Application Platform 之類的技術而構建的)集成到這些新的事件流架構中。

考慮以下示例： 某個網(wǎng)上商店擁有一個移動應用程序，該應用程序使用 Kafka 向企業(yè) Java 中實施的分布式支付系統(tǒng)發(fā)送支付請求數(shù)據(jù)。該解決方案必須絕對保證一次性完成支付請求的處理(避免多次向買家收費)。但是，在分布式系統(tǒng)中，故障是不可避免的，因此該解決方案需要采用穩(wěn)妥的方式來處理故障。

使用 Apache Kafka 實施消息傳遞

Apache Kafka 是一種用于事件流處理的分布式系統(tǒng)，廣泛應用于微服務架構和基于云的環(huán)境中。它在同一個平臺內(nèi)提供事件的消息傳遞、存儲和處理。

下圖展示了 Apache Kafka 組件的基本拓撲，其中包括通過 Kafka 集群基礎架構交換消息的生產(chǎn)者和使用者。

即使 Kafka 具有諸多優(yōu)勢，但 Kafka 仍面臨如下問題：

• 消息處理失敗時需要實施手動補償邏輯，這可能導致無法處理消息
• 不支持 XA 事務處理
• 確保在使用者應用程序中僅執(zhí)行一次交付處理
• 需要完成額外的開發(fā)和可維護性工作才能將其集成到企業(yè)解決方案中

要解決 Kafka 集成問題，您可以應用傳統(tǒng)消息傳遞拓撲概念，例如，事務日志、恢復日志和 XA 事務。 您可以實施基于 Java EE Connector Architecture (JCA) 的資源適配器。利用此 JCA 資源適配器，您可以為應用程序服務器提供 ACID 功能，以便進行 Kafka 消息處理。此 JCA 資源適配器隨后可提供與企業(yè) Java 應用程序的無縫 Kafka 集成。

實施 JCA 資源適配器

Java EE Connector Architecture 可定義一組可擴展且十分安全的事務性機制。您可以將 JCA 資源適配器安裝到兼容 Java EE 的任意應用程序服務器中，例如，IBM Websphere Application Server、IBM Business Process Manager、JBoss、WebSphere Liberty、Glassfish 或 Weblogic。

Java EE Connector Architecture 規(guī)范還提供了一組標準合約，用于支持企業(yè)應用程序與企業(yè)信息系統(tǒng)(如 Kafka)之間的通信。JCA 資源適配器可以插入到應用程序服務器，可通過處理所有系統(tǒng)級別的機制(事務、連接管理、崩潰恢復、錯誤跟蹤和日志記錄)來支持 Kafka 集成。JCA 資源適配器將對需要與之集成的企業(yè)應用程序隱藏所有 Kafka 通信邏輯。通過實施 JCA 資源適配器，企業(yè)應用程序提供商可以集中精力實施業(yè)務和演示邏輯，而無需擔心與 Kafka 集成相關的低級別邏輯。因此，JCA 資源適配器只需開發(fā)一次，即可供各種應用程序復用。

讓我們將它與網(wǎng)上商店支付場景聯(lián)系起來，觀察下圖，它展示了指定的解決方案系統(tǒng)上下文。

移動應用程序向 Kafka發(fā)送支付請求數(shù)據(jù)， 該Kafka 已通過資源適配器與企業(yè)支付應用程序進行了集成。此外，還可以使用此適配器向 Kafka 推送支付通知。此適配器會啟動 XA 事務，該事務將傳遞到企業(yè)支付應用程序和通知系統(tǒng)。因此，與支付請求處理相關的所有任務都將在同一個全局事務內(nèi)運行，并且同時完成或者同時失敗。除了從中讀取數(shù)據(jù)或向其中寫入數(shù)據(jù)的主題外，該設計還在 Kafka 上設置了重試、死信和事務日志主題。

現(xiàn)在，我們來更詳細地探索與移動應用程序往來的消息的處理過程。

傳入流

在我們的支付場景中，傳入流表示由網(wǎng)上商店移動應用程序發(fā)起的通信，該應用程序會向 Kafka 發(fā)送支付請求數(shù)據(jù)。資源適配器提供了 Kafka 連接，并向應用程序服務器上存在的消息端點異步傳遞消息。可使用 JCA 規(guī)范所定義的消息傳入流合約來實現(xiàn)這一點。

Kafka JCA 資源適配器會實施激活規(guī)范 JavaBean，其中包含一組用于端點激活配置的配置屬性。這些配置詳細信息將作為應用程序服務器配置的一部分來進行定義。

資源適配器會定期從傳入 Kafka 主題輪詢一批支付請求。成功完成數(shù)據(jù)輪詢后，它會迭代數(shù)據(jù)批次，并異步向端點實例傳遞消息。每個消息端點可能存在多個端點實例，因此能夠并行使用消息并提供高吞吐量。

Kafka 使用者偏移在安排消息送達后立即落實，從而避免了批次受阻的問題。這種設計是可行的，因為該資源適配器通過需要在 Kafka 上設置的重試、死信和事務日志主題來實施故障轉移過程。在我們的例子中，端點需要支持 XA 事務，并且需要在向端點發(fā)送數(shù)據(jù)之前創(chuàng)建事務上下文，從而提供原子消息使用。

如果應用程序服務器異常終止了事務，那么由端點實例執(zhí)行的所有工作都應回滾，并且消息應轉發(fā)到 Kafka 重試主題。

適配器使用來自 Kafka 重試主題的消息，并對其進行重新處理。超出已配置的消息處理重試次數(shù)后，該適配器會將此消息傳遞到 Kafka 死信主題。發(fā)送到死信主題的消息包含有價值的業(yè)務數(shù)據(jù)，因此監(jiān)視該主題至關重要。

傳出流

傳出流表示由企業(yè)應用程序發(fā)起的 Kafka 通信。在我們的例子中，這是用于向移動應用程序發(fā)送支付確認的通知系統(tǒng)。JCA 規(guī)范定義了一個連接管理合約，可讓應用程序服務器合并 Kafka 連接，從而提供支持大量客戶端的可擴展環(huán)境。

Kafka 傳出連接配置詳細信息是使用 Managed Connection Factory JavaBean 進行定義的。利用這些配置詳細信息，管理員和開發(fā)者可使用適配器來配置 Kafka 生產(chǎn)者，并決定所需的功能，例如，可靠性、可用性、吞吐量、延遲和事務支持。這些配置詳細信息將作為應用程序服務器配置的一部分來進行定義。

Kafka JCA 資源適配器將公開用于實施公共客戶端接口 (CCI) 和 Java 消息服務 (JMS) 接口的 Kafka Connection Factory 和 Kafka Connection。應用程序組件會使用 Java 命名和目錄接口 (JNDI) 名稱來查找連接工廠。成功獲取連接工廠后，應用程序會使用它來獲取連接，以便訪問 Kafka。這樣，您就可以為通知系統(tǒng)應用程序無縫添加 Kafka 集成，該應用程序當前將數(shù)據(jù)發(fā)送到 JMS 消息傳遞提供程序(如 IBM MQ 或 Active MQ)。

資源適配器傳出流會封裝低級別的 Kafka 通信邏輯，并提供：

• 連接合并
• 使用 Kafka 事務性機制來保證僅傳遞一次
• 采用穩(wěn)妥的方式來識別、記錄和處理 Kafka 故障
• 實施 XA 事務，從而在分布式系統(tǒng)中通過 Kafka 提供可靠的消息處理

要在傳出流中管理事務，Kafka 資源適配器可使用由 JCA 規(guī)范定義的事務管理合約。

在我們的例子中，連接工廠需要設置為支持 XA 事務，該適配器需要在客戶端獲取連接時啟動 Kafka 事務。無論應用程序服務器何時回滾 Kafka 事務，該事務都會異常終止。如果發(fā)生 XA 事務落實，那么事務管理器會在正在運行的事務所使用的所有資源上執(zhí)行兩階段落實協(xié)議。這可保證對受管資源的所有讀寫訪問權要么全部落實，要么全部回滾。

最后，該資源適配器會通過向 Kafka 事務日志主題寫入事務數(shù)據(jù)來跟蹤正在運行的事務。寫入事務日志主題的數(shù)據(jù)將用于崩潰恢復處理，以便在分布式系統(tǒng)中提供可靠的消息處理。

結束語

Kafka JCA 適配器的設計方法為標準企業(yè) Java 解決方案提供了 JMS 與 Kafka 事件處理平臺的"即插即用式"集成。此設計支持將 Kafka 與現(xiàn)有企業(yè)應用程序無縫集成，而無需實施補償邏輯。通過該適配器，應用程序服務器還可以提供企業(yè)應用程序所依賴的基礎架構和運行時環(huán)境，以用于建立 Kafka 連接并執(zhí)行事務管理。