世界已經(jīng)邁進(jìn)"移動(dòng)"時(shí)代，現(xiàn)在應(yīng)用程序必須能夠?qū)崟r(shí)提供數(shù)據(jù)，這不僅包括數(shù)據(jù)庫(kù)表中存儲(chǔ)的重要最終結(jié)果，還包括用戶使用應(yīng)用程序時(shí)執(zhí)行的所有操作。任何可用信息，例如，用戶點(diǎn)擊量、日志數(shù)據(jù)或傳感器數(shù)據(jù)都可用于改善用戶體驗(yàn)、生成報(bào)告、向機(jī)器學(xué)習(xí)系統(tǒng)提供數(shù)據(jù)，等等。現(xiàn)如今，開(kāi)發(fā)者必須關(guān)注基于實(shí)時(shí)事件流的系統(tǒng)。
下圖展示了基于事件流處理的架構(gòu)示例。
Apache Kafka 已成為構(gòu)建基于事件的高度可擴(kuò)展系統(tǒng)的首選解決方案。Kafka 可為事件流平臺(tái)提供快速升級(jí)換代的功能，以供開(kāi)發(fā)者在現(xiàn)代化業(yè)務(wù)解決方案中使用。然而，開(kāi)發(fā)者通常需要將現(xiàn)有 Java EE 業(yè)務(wù)解決方案(基于 IBM MQ 或 IBM WebSphere Application Platform 之類的技術(shù)而構(gòu)建的)集成到這些新的事件流架構(gòu)中。
考慮以下示例： 某個(gè)網(wǎng)上商店擁有一個(gè)移動(dòng)應(yīng)用程序，該應(yīng)用程序使用 Kafka 向企業(yè) Java 中實(shí)施的分布式支付系統(tǒng)發(fā)送支付請(qǐng)求數(shù)據(jù)。該解決方案必須絕對(duì)保證一次性完成支付請(qǐng)求的處理(避免多次向買家收費(fèi))。但是，在分布式系統(tǒng)中，故障是不可避免的，因此該解決方案需要采用穩(wěn)妥的方式來(lái)處理故障。
使用 Apache Kafka 實(shí)施消息傳遞
Apache Kafka 是一種用于事件流處理的分布式系統(tǒng)，廣泛應(yīng)用于微服務(wù)架構(gòu)和基于云的環(huán)境中。它在同一個(gè)平臺(tái)內(nèi)提供事件的消息傳遞、存儲(chǔ)和處理。
下圖展示了 Apache Kafka 組件的基本拓?fù)?，其中包括通過(guò) Kafka 集群基礎(chǔ)架構(gòu)交換消息的生產(chǎn)者和使用者。
即使 Kafka 具有諸多優(yōu)勢(shì)，但 Kafka 仍面臨如下問(wèn)題：
- 消息處理失敗時(shí)需要實(shí)施手動(dòng)補(bǔ)償邏輯，這可能導(dǎo)致無(wú)法處理消息
- 不支持 XA 事務(wù)處理
- 確保在使用者應(yīng)用程序中僅執(zhí)行一次交付處理
- 需要完成額外的開(kāi)發(fā)和可維護(hù)性工作才能將其集成到企業(yè)解決方案中
要解決 Kafka 集成問(wèn)題，您可以應(yīng)用傳統(tǒng)消息傳遞拓?fù)涓拍睿?，事?wù)日志、恢復(fù)日志和 XA 事務(wù)。 您可以實(shí)施基于 Java EE Connector Architecture (JCA) 的資源適配器。利用此 JCA 資源適配器，您可以為應(yīng)用程序服務(wù)器提供 ACID 功能，以便進(jìn)行 Kafka 消息處理。此 JCA 資源適配器隨后可提供與企業(yè) Java 應(yīng)用程序的無(wú)縫 Kafka 集成。
實(shí)施 JCA 資源適配器
Java EE Connector Architecture 可定義一組可擴(kuò)展且十分安全的事務(wù)性機(jī)制。您可以將 JCA 資源適配器安裝到兼容 Java EE 的任意應(yīng)用程序服務(wù)器中，例如，IBM Websphere Application Server、IBM Business Process Manager、JBoss、WebSphere Liberty、Glassfish 或 Weblogic。
Java EE Connector Architecture 規(guī)范還提供了一組標(biāo)準(zhǔn)合約，用于支持企業(yè)應(yīng)用程序與企業(yè)信息系統(tǒng)(如 Kafka)之間的通信。JCA 資源適配器可以插入到應(yīng)用程序服務(wù)器，可通過(guò)處理所有系統(tǒng)級(jí)別的機(jī)制(事務(wù)、連接管理、崩潰恢復(fù)、錯(cuò)誤跟蹤和日志記錄)來(lái)支持 Kafka 集成。JCA 資源適配器將對(duì)需要與之集成的企業(yè)應(yīng)用程序隱藏所有 Kafka 通信邏輯。通過(guò)實(shí)施 JCA 資源適配器，企業(yè)應(yīng)用程序提供商可以集中精力實(shí)施業(yè)務(wù)和演示邏輯，而無(wú)需擔(dān)心與 Kafka 集成相關(guān)的低級(jí)別邏輯。因此，JCA 資源適配器只需開(kāi)發(fā)一次，即可供各種應(yīng)用程序復(fù)用。
讓我們將它與網(wǎng)上商店支付場(chǎng)景聯(lián)系起來(lái)，觀察下圖，它展示了指定的解決方案系統(tǒng)上下文。
移動(dòng)應(yīng)用程序向 Kafka發(fā)送支付請(qǐng)求數(shù)據(jù)， 該Kafka 已通過(guò)資源適配器與企業(yè)支付應(yīng)用程序進(jìn)行了集成。此外，還可以使用此適配器向 Kafka 推送支付通知。此適配器會(huì)啟動(dòng) XA 事務(wù)，該事務(wù)將傳遞到企業(yè)支付應(yīng)用程序和通知系統(tǒng)。因此，與支付請(qǐng)求處理相關(guān)的所有任務(wù)都將在同一個(gè)全局事務(wù)內(nèi)運(yùn)行，并且同時(shí)完成或者同時(shí)失敗。除了從中讀取數(shù)據(jù)或向其中寫入數(shù)據(jù)的主題外，該設(shè)計(jì)還在 Kafka 上設(shè)置了重試、死信和事務(wù)日志主題。
現(xiàn)在，我們來(lái)更詳細(xì)地探索與移動(dòng)應(yīng)用程序往來(lái)的消息的處理過(guò)程。
傳入流
在我們的支付場(chǎng)景中，傳入流表示由網(wǎng)上商店移動(dòng)應(yīng)用程序發(fā)起的通信，該應(yīng)用程序會(huì)向 Kafka 發(fā)送支付請(qǐng)求數(shù)據(jù)。資源適配器提供了 Kafka 連接，并向應(yīng)用程序服務(wù)器上存在的消息端點(diǎn)異步傳遞消息?？墒褂?JCA 規(guī)范所定義的消息傳入流合約來(lái)實(shí)現(xiàn)這一點(diǎn)。
Kafka JCA 資源適配器會(huì)實(shí)施激活規(guī)范 JavaBean，其中包含一組用于端點(diǎn)激活配置的配置屬性。這些配置詳細(xì)信息將作為應(yīng)用程序服務(wù)器配置的一部分來(lái)進(jìn)行定義。
資源適配器會(huì)定期從傳入 Kafka 主題輪詢一批支付請(qǐng)求。成功完成數(shù)據(jù)輪詢后，它會(huì)迭代數(shù)據(jù)批次，并異步向端點(diǎn)實(shí)例傳遞消息。每個(gè)消息端點(diǎn)可能存在多個(gè)端點(diǎn)實(shí)例，因此能夠并行使用消息并提供高吞吐量。
Kafka 使用者偏移在安排消息送達(dá)后立即落實(shí)，從而避免了批次受阻的問(wèn)題。這種設(shè)計(jì)是可行的，因?yàn)樵撡Y源適配器通過(guò)需要在 Kafka 上設(shè)置的重試、死信和事務(wù)日志主題來(lái)實(shí)施故障轉(zhuǎn)移過(guò)程。在我們的例子中，端點(diǎn)需要支持 XA 事務(wù)，并且需要在向端點(diǎn)發(fā)送數(shù)據(jù)之前創(chuàng)建事務(wù)上下文，從而提供原子消息使用。
如果應(yīng)用程序服務(wù)器異常終止了事務(wù)，那么由端點(diǎn)實(shí)例執(zhí)行的所有工作都應(yīng)回滾，并且消息應(yīng)轉(zhuǎn)發(fā)到 Kafka 重試主題。
適配器使用來(lái)自 Kafka 重試主題的消息，并對(duì)其進(jìn)行重新處理。超出已配置的消息處理重試次數(shù)后，該適配器會(huì)將此消息傳遞到 Kafka 死信主題。發(fā)送到死信主題的消息包含有價(jià)值的業(yè)務(wù)數(shù)據(jù)，因此監(jiān)視該主題至關(guān)重要。
傳出流
傳出流表示由企業(yè)應(yīng)用程序發(fā)起的 Kafka 通信。在我們的例子中，這是用于向移動(dòng)應(yīng)用程序發(fā)送支付確認(rèn)的通知系統(tǒng)。JCA 規(guī)范定義了一個(gè)連接管理合約，可讓應(yīng)用程序服務(wù)器合并 Kafka 連接，從而提供支持大量客戶端的可擴(kuò)展環(huán)境。
Kafka 傳出連接配置詳細(xì)信息是使用 Managed Connection Factory JavaBean 進(jìn)行定義的。利用這些配置詳細(xì)信息，管理員和開(kāi)發(fā)者可使用適配器來(lái)配置 Kafka 生產(chǎn)者，并決定所需的功能，例如，可靠性、可用性、吞吐量、延遲和事務(wù)支持。這些配置詳細(xì)信息將作為應(yīng)用程序服務(wù)器配置的一部分來(lái)進(jìn)行定義。
Kafka JCA 資源適配器將公開(kāi)用于實(shí)施公共客戶端接口 (CCI) 和 Java 消息服務(wù) (JMS) 接口的 Kafka Connection Factory 和 Kafka Connection。應(yīng)用程序組件會(huì)使用 Java 命名和目錄接口 (JNDI) 名稱來(lái)查找連接工廠。成功獲取連接工廠后，應(yīng)用程序會(huì)使用它來(lái)獲取連接，以便訪問(wèn) Kafka。這樣，您就可以為通知系統(tǒng)應(yīng)用程序無(wú)縫添加 Kafka 集成，該應(yīng)用程序當(dāng)前將數(shù)據(jù)發(fā)送到 JMS 消息傳遞提供程序(如 IBM MQ 或 Active MQ)。
資源適配器傳出流會(huì)封裝低級(jí)別的 Kafka 通信邏輯，并提供：
- 連接合并
- 使用 Kafka 事務(wù)性機(jī)制來(lái)保證僅傳遞一次
- 采用穩(wěn)妥的方式來(lái)識(shí)別、記錄和處理 Kafka 故障
- 實(shí)施 XA 事務(wù)，從而在分布式系統(tǒng)中通過(guò) Kafka 提供可靠的消息處理
要在傳出流中管理事務(wù)，Kafka 資源適配器可使用由 JCA 規(guī)范定義的事務(wù)管理合約。
在我們的例子中，連接工廠需要設(shè)置為支持 XA 事務(wù)，該適配器需要在客戶端獲取連接時(shí)啟動(dòng) Kafka 事務(wù)。無(wú)論應(yīng)用程序服務(wù)器何時(shí)回滾 Kafka 事務(wù)，該事務(wù)都會(huì)異常終止。如果發(fā)生 XA 事務(wù)落實(shí)，那么事務(wù)管理器會(huì)在正在運(yùn)行的事務(wù)所使用的所有資源上執(zhí)行兩階段落實(shí)協(xié)議。這可保證對(duì)受管資源的所有讀寫訪問(wèn)權(quán)要么全部落實(shí)，要么全部回滾。
最后，該資源適配器會(huì)通過(guò)向 Kafka 事務(wù)日志主題寫入事務(wù)數(shù)據(jù)來(lái)跟蹤正在運(yùn)行的事務(wù)。寫入事務(wù)日志主題的數(shù)據(jù)將用于崩潰恢復(fù)處理，以便在分布式系統(tǒng)中提供可靠的消息處理。
結(jié)束語(yǔ)
Kafka JCA 適配器的設(shè)計(jì)方法為標(biāo)準(zhǔn)企業(yè) Java 解決方案提供了 JMS 與 Kafka 事件處理平臺(tái)的"即插即用式"集成。此設(shè)計(jì)支持將 Kafka 與現(xiàn)有企業(yè)應(yīng)用程序無(wú)縫集成，而無(wú)需實(shí)施補(bǔ)償邏輯。通過(guò)該適配器，應(yīng)用程序服務(wù)器還可以提供企業(yè)應(yīng)用程序所依賴的基礎(chǔ)架構(gòu)和運(yùn)行時(shí)環(huán)境，以用于建立 Kafka 連接并執(zhí)行事務(wù)管理。
【責(zé)任編輯：未麗燕 TEL：（010）68476606】