一種基于Netty 的異步處理方法

曹承飛

（上海浦東發展銀行總行信息科技部，上海 200000）

0 引言

隨著金融科技的快速發展，IT 系統的不斷建設業務系統間的交互變得愈加頻繁，業務系統間接口數據量呈幾何級數增長。隨著業務系統增多，對交互處理能力也越來越高。如何提供一種強處理、高并發和高穩定性的電子化聯通處理方法顯得尤為重要。

傳統的聯通模式為各業務系統間建立連接，一般采用發送請求并同步等待應答的模式。該處理方式存在以下問題：①會導致系統間連接異常復雜且不易管理。②會影響系統間的交互響應速度。③同步的處理模式占用較多的系統資源，對處理性能帶來一定的影響。

Netty 是一種異步事件驅動的網絡應用框架，擁有簡單而強大的線程模型，具有高吞吐、低延遲，低能耗的特性等優點。本文提供一種基于Netty 的消息異步處理方法，包括：請求處理模塊、授權處理模塊、報文處理模塊、消息處理模塊、流水信息處理模塊以及日志處理模塊，可以提高系統間交互的響應速度以及處理性能。

1 異步通信處理方法

本文提出一種基于Netty 的異步聯通處理方法[1-4]，該方法可替代現有的各系統間相互連接模式，實現滿足不同實時性需求的高效、高可用性的消息交互，實現“一點觸發，多點聯動”。

該方法在Redis 等基礎設施上建立，包括：接入處理模塊、授權處理模塊、報文處理模塊、消息處理模塊、流水處理模塊、日志處理模塊，具體功能架構如圖1 所示。

圖1 異步處理功能架構

請求處理模塊用于接收多個上游系統通過不同通道發送的請求消息，并將請求消息對應的下游系統發送來的應答消息寫入對應的通道中，以將應答消息異步返回至對應的上游系統。

授權處理模塊用于判斷所述上游系統是否注冊，若注冊，則允許所述上游系統接入。

報文處理模塊用于對報文進行格式轉化、加密解密、特殊處理及對報文池進行管理。

消息處理模塊用于計算消息路由、消息發送、消息清理、消息緩存、消息重發及消息異步中轉。

流水信息處理模塊用于將流水信息寫入流水隊列，并異步從所述流水隊列中讀取流水信息，將讀取的流水信息進行登記。

日志處理模塊用于將日志信息寫入日志隊列，并異步從所述日志隊列中讀取日志信息，將讀取的日志信息進行登記。

2 異步通信處理具體實現方式

本章節結合圖1 對異步通信處理方法的實現方式做進一步的詳細闡述。

如圖1 所示，該方法包括請求處理模塊110、授權處理模塊120、報文處理模塊130、消息處理模塊140、流水信息處理模塊150 以及日志處理模塊160。

請求處理模塊110 用于接收多個上游系統通過不同通道發送的請求消息，并將請求消息對應的下游系統發送來的應答消息寫入對應的通道中，以將應答消息異步返回至對應的上游系統。

其中，請求信息中攜帶有報文。報文包括報文頭和報文體兩部分，報文體由屬性域和數據域兩部分組成。屬性域用于存放與數據域中的應用數據有關的各項屬性，數據域存放各應用系統所需傳輸的業務數據，采用文本格式或者JSON 格式。報文體的屬性域包括有9 項數據元，如表1 所示。

表1 報文體屬性域定義

具體的，請求處理模塊110 包括請求接收單元、應答選擇單元、應答發送單元。其中，請求接收單元用于接收多個上游系統通過不同通道發送的請求消息，并建立請求消息的消息標識與通道標識的對應關系，將對應關系進行緩存。應答選擇單元用于根據下游系統發送來的應答消息的消息標識和對應關系選擇通道。應答發送單元用于將應答消息寫入選擇的通道中，以通過選擇的通道將應答消息返回至上游系統。

其中，請求消息的消息標識與應答消息的消息標識相同。消息標識可以是上表中的“MSGID”數據元。

本實施案例中，請求接收單元基于Netty 完成[5-7]，通過不同的通道（channel）完成請求消息的接收，在收到請求消息后將請求消息中的MSGID 及其對應channel 的標識（ctx）存入緩存區中，以便后續進行應答選擇處理。應答選擇單元通過使用應答消息中的MSGID 匹配到對應channel 的ctx，以便將將應答消息寫入原channel，實現消息的異步原路返回。應答發送單元將應答消息寫入所選擇的channel 中，以將應答消息返回至上游系統。

授權處理模塊120 用于判斷上游系統是否注冊，若注冊，則允許上游系統接入。

本案例中，只有注冊登記的上游系統才允許接入。

報文處理模塊130 用于對報文進行格式轉化、加密解密、特殊處理及對報文池進行管理。

具體的，報文處理模塊130 包括請求報文轉化單元、應答報文轉化單元、報文加解密單元、特殊報文處理單元及報文池管理單元。

其中，請求報文轉化單元用于將請求報文的格式根據配置信息轉化成系統內部格式。應答報文轉化單元用于將應答報文從內部報文格式轉化為外部系統可識別的格式。報文加解密單元用于對請求報文進行加密，對應答報文進行解密。特殊報文處理單元用于對報文中的設定關鍵字進行替換或者刪除。報文池管理單元用于對系統內部報文進行管理，以便報文在系統內部流轉。

消息處理模塊140 包括消息路由計算單元、消息發送單元、消息清理單元及消息異步中轉單元。

其中，消息路由計算單元用于根據報文中的發送系統信息、接收系統信息及設定規則計算路由信息，路由信息包括目標系統信息。消息發送單元用于將消息發送至確定的目標系統。消息清理單元用于將隊列中超過設定生存時長的消息進行清理。消息異步中轉單元用于當判斷下游系統返回的應答消息為中轉類型時，提取應答消息的設定關鍵字，并從設定緩存中獲取中轉配置規則，根據設定關鍵字和中轉配置規則確定目標系統，最后將應答消息發送至目標系統。

本案例中，消息路由計算單元首先獲取消息的屬性信息以及路由規則，基于屬性信息和路由規則計算目標系統，如果計算成功，則將消息發送至目標系統，若計算失敗，則進行路由失敗反饋。

本案例中，消息發送單元基于Netty 協議將消息發送至目標系統[8]。本實施例中消息發送單元的原理如圖2 所示，接收到請求消息后，首先對接入點權限（上游系統權限）進行檢查，若成功，則進行消息路由計算，并判斷路由是否成功，若成功，則判斷是否需要中轉，若不需要中轉，則對消息進行處理，并將處理后的消息進行發送。

圖2 消息發送單元原理

本案例中，消息異步中轉的場景如下：上游系統A發送請求消息給下游系統B，系統B 將應答消息返回至異步處理系統，異步處理系統判斷為中轉類型消息，并從Redis 中讀取中轉配置規則，根據中轉規則配置，獲取系統B 返回報文中的關鍵字；根據關鍵字的配置，計算中轉的目標系統；將消息打包發送至目標系統。

本案例中，消息清理單元用于將隊列中超過設定生存時長的消息確定為死信消息，獲取死信消息的清理規則，按照請求規則對死信消息進行清理。清理規則包括重發和刪除。當清理規則為重發時，判斷死信消息的重發次數是否超過設定閾值，若超過，則將死信消息刪除。本實施例中消息清理單元的原理如圖3 所示，首先獲取死信消息，然后確定死信消息的類型，然后根據類型確定清理屬性，基于清理屬性確定清理類型；若清理類型為重發，則獲取目標系統，并對重發頻次進行控制，將消息打包后重發至目標系統；若清理類型為刪除，則將死信消息刪除。

圖3 消息清理單元原理

流水信息處理模塊150 用于將流水信息寫入流水隊列，并異步從流水隊列中讀取流水信息，將讀取的流水信息進行登記，為后續的記錄查詢、記錄核對、重發恢復等提供了重要的保障。

具體的，流水信息處理模塊獲取流水信息及流水登記規則，根據流水登記規則計算流水隊列，將流水信息寫入計算出的流水隊列中；異步從流水隊列中讀取流水信息，將讀取的流水信息進行等級。

日志處理模塊160 用于將日志信息寫入日志隊列，并異步從日志隊列中讀取日志信息，將讀取的日志信息進行登記。

3 結語

本文提供一種基于Netty 的異步處理方法，大大降低了系統間的耦合度，提升了處理性能。雖然本文對該方法的具體實現方式作了較為詳細的說明，但在實際應用中不僅僅限于以上所述模塊內容，在不脫離本文所述方法構思的情況下，還可以包括更多其他等效模塊，以應對更加復雜的應用場景。