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

曹承飛

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

0 引言

隨著金融科技的快速發(fā)展，IT 系統(tǒng)的不斷建設(shè)業(yè)務(wù)系統(tǒng)間的交互變得愈加頻繁，業(yè)務(wù)系統(tǒng)間接口數(shù)據(jù)量呈幾何級(jí)數(shù)增長。隨著業(yè)務(wù)系統(tǒng)增多，對(duì)交互處理能力也越來越高。如何提供一種強(qiáng)處理、高并發(fā)和高穩(wěn)定性的電子化聯(lián)通處理方法顯得尤為重要。

傳統(tǒng)的聯(lián)通模式為各業(yè)務(wù)系統(tǒng)間建立連接，一般采用發(fā)送請(qǐng)求并同步等待應(yīng)答的模式。該處理方式存在以下問題：①會(huì)導(dǎo)致系統(tǒng)間連接異常復(fù)雜且不易管理。②會(huì)影響系統(tǒng)間的交互響應(yīng)速度。③同步的處理模式占用較多的系統(tǒng)資源，對(duì)處理性能帶來一定的影響。

Netty 是一種異步事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用框架，擁有簡單而強(qiáng)大的線程模型，具有高吞吐、低延遲，低能耗的特性等優(yōu)點(diǎn)。本文提供一種基于Netty 的消息異步處理方法，包括：請(qǐng)求處理模塊、授權(quán)處理模塊、報(bào)文處理模塊、消息處理模塊、流水信息處理模塊以及日志處理模塊，可以提高系統(tǒng)間交互的響應(yīng)速度以及處理性能。

1 異步通信處理方法

本文提出一種基于Netty 的異步聯(lián)通處理方法[1-4]，該方法可替代現(xiàn)有的各系統(tǒng)間相互連接模式，實(shí)現(xiàn)滿足不同實(shí)時(shí)性需求的高效、高可用性的消息交互，實(shí)現(xiàn)“一點(diǎn)觸發(fā)，多點(diǎn)聯(lián)動(dòng)”。

該方法在Redis 等基礎(chǔ)設(shè)施上建立，包括：接入處理模塊、授權(quán)處理模塊、報(bào)文處理模塊、消息處理模塊、流水處理模塊、日志處理模塊，具體功能架構(gòu)如圖1 所示。

圖1 異步處理功能架構(gòu)

請(qǐng)求處理模塊用于接收多個(gè)上游系統(tǒng)通過不同通道發(fā)送的請(qǐng)求消息，并將請(qǐng)求消息對(duì)應(yīng)的下游系統(tǒng)發(fā)送來的應(yīng)答消息寫入對(duì)應(yīng)的通道中，以將應(yīng)答消息異步返回至對(duì)應(yīng)的上游系統(tǒng)。

授權(quán)處理模塊用于判斷所述上游系統(tǒng)是否注冊(cè)，若注冊(cè)，則允許所述上游系統(tǒng)接入。

報(bào)文處理模塊用于對(duì)報(bào)文進(jìn)行格式轉(zhuǎn)化、加密解密、特殊處理及對(duì)報(bào)文池進(jìn)行管理。

消息處理模塊用于計(jì)算消息路由、消息發(fā)送、消息清理、消息緩存、消息重發(fā)及消息異步中轉(zhuǎn)。

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

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

2 異步通信處理具體實(shí)現(xiàn)方式

本章節(jié)結(jié)合圖1 對(duì)異步通信處理方法的實(shí)現(xiàn)方式做進(jìn)一步的詳細(xì)闡述。

如圖1 所示，該方法包括請(qǐng)求處理模塊110、授權(quán)處理模塊120、報(bào)文處理模塊130、消息處理模塊140、流水信息處理模塊150 以及日志處理模塊160。

請(qǐng)求處理模塊110 用于接收多個(gè)上游系統(tǒng)通過不同通道發(fā)送的請(qǐng)求消息，并將請(qǐng)求消息對(duì)應(yīng)的下游系統(tǒng)發(fā)送來的應(yīng)答消息寫入對(duì)應(yīng)的通道中，以將應(yīng)答消息異步返回至對(duì)應(yīng)的上游系統(tǒng)。

其中，請(qǐng)求信息中攜帶有報(bào)文。報(bào)文包括報(bào)文頭和報(bào)文體兩部分，報(bào)文體由屬性域和數(shù)據(jù)域兩部分組成。屬性域用于存放與數(shù)據(jù)域中的應(yīng)用數(shù)據(jù)有關(guān)的各項(xiàng)屬性，數(shù)據(jù)域存放各應(yīng)用系統(tǒng)所需傳輸?shù)臉I(yè)務(wù)數(shù)據(jù)，采用文本格式或者JSON 格式。報(bào)文體的屬性域包括有9 項(xiàng)數(shù)據(jù)元，如表1 所示。

表1 報(bào)文體屬性域定義

具體的，請(qǐng)求處理模塊110 包括請(qǐng)求接收單元、應(yīng)答選擇單元、應(yīng)答發(fā)送單元。其中，請(qǐng)求接收單元用于接收多個(gè)上游系統(tǒng)通過不同通道發(fā)送的請(qǐng)求消息，并建立請(qǐng)求消息的消息標(biāo)識(shí)與通道標(biāo)識(shí)的對(duì)應(yīng)關(guān)系，將對(duì)應(yīng)關(guān)系進(jìn)行緩存。應(yīng)答選擇單元用于根據(jù)下游系統(tǒng)發(fā)送來的應(yīng)答消息的消息標(biāo)識(shí)和對(duì)應(yīng)關(guān)系選擇通道。應(yīng)答發(fā)送單元用于將應(yīng)答消息寫入選擇的通道中，以通過選擇的通道將應(yīng)答消息返回至上游系統(tǒng)。

其中，請(qǐng)求消息的消息標(biāo)識(shí)與應(yīng)答消息的消息標(biāo)識(shí)相同。消息標(biāo)識(shí)可以是上表中的“MSGID”數(shù)據(jù)元。

本實(shí)施案例中，請(qǐng)求接收單元基于Netty 完成[5-7]，通過不同的通道（channel）完成請(qǐng)求消息的接收，在收到請(qǐng)求消息后將請(qǐng)求消息中的MSGID 及其對(duì)應(yīng)channel 的標(biāo)識(shí)（ctx）存入緩存區(qū)中，以便后續(xù)進(jìn)行應(yīng)答選擇處理。應(yīng)答選擇單元通過使用應(yīng)答消息中的MSGID 匹配到對(duì)應(yīng)channel 的ctx，以便將將應(yīng)答消息寫入原channel，實(shí)現(xiàn)消息的異步原路返回。應(yīng)答發(fā)送單元將應(yīng)答消息寫入所選擇的channel 中，以將應(yīng)答消息返回至上游系統(tǒng)。

授權(quán)處理模塊120 用于判斷上游系統(tǒng)是否注冊(cè)，若注冊(cè)，則允許上游系統(tǒng)接入。

本案例中，只有注冊(cè)登記的上游系統(tǒng)才允許接入。

報(bào)文處理模塊130 用于對(duì)報(bào)文進(jìn)行格式轉(zhuǎn)化、加密解密、特殊處理及對(duì)報(bào)文池進(jìn)行管理。

具體的，報(bào)文處理模塊130 包括請(qǐng)求報(bào)文轉(zhuǎn)化單元、應(yīng)答報(bào)文轉(zhuǎn)化單元、報(bào)文加解密單元、特殊報(bào)文處理單元及報(bào)文池管理單元。

其中，請(qǐng)求報(bào)文轉(zhuǎn)化單元用于將請(qǐng)求報(bào)文的格式根據(jù)配置信息轉(zhuǎn)化成系統(tǒng)內(nèi)部格式。應(yīng)答報(bào)文轉(zhuǎn)化單元用于將應(yīng)答報(bào)文從內(nèi)部報(bào)文格式轉(zhuǎn)化為外部系統(tǒng)可識(shí)別的格式。報(bào)文加解密單元用于對(duì)請(qǐng)求報(bào)文進(jìn)行加密，對(duì)應(yīng)答報(bào)文進(jìn)行解密。特殊報(bào)文處理單元用于對(duì)報(bào)文中的設(shè)定關(guān)鍵字進(jìn)行替換或者刪除。報(bào)文池管理單元用于對(duì)系統(tǒng)內(nèi)部報(bào)文進(jìn)行管理，以便報(bào)文在系統(tǒng)內(nèi)部流轉(zhuǎn)。

消息處理模塊140 包括消息路由計(jì)算單元、消息發(fā)送單元、消息清理單元及消息異步中轉(zhuǎn)單元。

其中，消息路由計(jì)算單元用于根據(jù)報(bào)文中的發(fā)送系統(tǒng)信息、接收系統(tǒng)信息及設(shè)定規(guī)則計(jì)算路由信息，路由信息包括目標(biāo)系統(tǒng)信息。消息發(fā)送單元用于將消息發(fā)送至確定的目標(biāo)系統(tǒng)。消息清理單元用于將隊(duì)列中超過設(shè)定生存時(shí)長的消息進(jìn)行清理。消息異步中轉(zhuǎn)單元用于當(dāng)判斷下游系統(tǒng)返回的應(yīng)答消息為中轉(zhuǎn)類型時(shí)，提取應(yīng)答消息的設(shè)定關(guān)鍵字，并從設(shè)定緩存中獲取中轉(zhuǎn)配置規(guī)則，根據(jù)設(shè)定關(guān)鍵字和中轉(zhuǎn)配置規(guī)則確定目標(biāo)系統(tǒng)，最后將應(yīng)答消息發(fā)送至目標(biāo)系統(tǒng)。

本案例中，消息路由計(jì)算單元首先獲取消息的屬性信息以及路由規(guī)則，基于屬性信息和路由規(guī)則計(jì)算目標(biāo)系統(tǒng)，如果計(jì)算成功，則將消息發(fā)送至目標(biāo)系統(tǒng)，若計(jì)算失敗，則進(jìn)行路由失敗反饋。

本案例中，消息發(fā)送單元基于Netty 協(xié)議將消息發(fā)送至目標(biāo)系統(tǒng)[8]。本實(shí)施例中消息發(fā)送單元的原理如圖2 所示，接收到請(qǐng)求消息后，首先對(duì)接入點(diǎn)權(quán)限（上游系統(tǒng)權(quán)限）進(jìn)行檢查，若成功，則進(jìn)行消息路由計(jì)算，并判斷路由是否成功，若成功，則判斷是否需要中轉(zhuǎn)，若不需要中轉(zhuǎn)，則對(duì)消息進(jìn)行處理，并將處理后的消息進(jìn)行發(fā)送。

圖2 消息發(fā)送單元原理

本案例中，消息異步中轉(zhuǎn)的場景如下：上游系統(tǒng)A發(fā)送請(qǐng)求消息給下游系統(tǒng)B，系統(tǒng)B 將應(yīng)答消息返回至異步處理系統(tǒng)，異步處理系統(tǒng)判斷為中轉(zhuǎn)類型消息，并從Redis 中讀取中轉(zhuǎn)配置規(guī)則，根據(jù)中轉(zhuǎn)規(guī)則配置，獲取系統(tǒng)B 返回報(bào)文中的關(guān)鍵字；根據(jù)關(guān)鍵字的配置，計(jì)算中轉(zhuǎn)的目標(biāo)系統(tǒng)；將消息打包發(fā)送至目標(biāo)系統(tǒng)。

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

圖3 消息清理單元原理

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

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

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

3 結(jié)語

本文提供一種基于Netty 的異步處理方法，大大降低了系統(tǒng)間的耦合度，提升了處理性能。雖然本文對(duì)該方法的具體實(shí)現(xiàn)方式作了較為詳細(xì)的說明，但在實(shí)際應(yīng)用中不僅僅限于以上所述模塊內(nèi)容，在不脫離本文所述方法構(gòu)思的情況下，還可以包括更多其他等效模塊，以應(yīng)對(duì)更加復(fù)雜的應(yīng)用場景。