胡佳

（中國聯(lián)合網(wǎng)絡(luò)通信有限公司廣東省分公司，廣東 廣州 510320）

1 引言

隨著移動(dòng)互聯(lián)網(wǎng)和手機(jī)的不斷發(fā)展和普及，電信行業(yè)的數(shù)據(jù)業(yè)務(wù)量呈爆發(fā)式增長。針對如何處理如此大量的數(shù)據(jù)，在現(xiàn)有的解決方案中最普遍的就是搭建一個(gè)分布式的消息隊(duì)列進(jìn)行數(shù)據(jù)存儲(chǔ)，數(shù)據(jù)存儲(chǔ)在消息隊(duì)列中等待被處理，這樣可以實(shí)現(xiàn)消息的不遺漏處理。

現(xiàn)有的按照普通消息隊(duì)列進(jìn)行消費(fèi)的處理方式存在明顯的缺點(diǎn)。數(shù)據(jù)業(yè)務(wù)可細(xì)分出緊急業(yè)務(wù)，正常業(yè)務(wù)，低優(yōu)先級業(yè)務(wù)等，當(dāng)大量的低優(yōu)先級或者普通業(yè)務(wù)堆積時(shí)，在現(xiàn)有的普通的消息隊(duì)列中，需要按照時(shí)間順序來一個(gè)一個(gè)處理，緊急業(yè)務(wù)不能得到優(yōu)先處理，導(dǎo)致緊急業(yè)務(wù)被貽誤，從而造成不必要的損失。

本研究的目的在于提供一種基于消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用系統(tǒng)，解決現(xiàn)有技術(shù)存在的無法按照優(yōu)先級處理訂單的問題。通過增添消息隊(duì)列topic個(gè)數(shù)并且賦予每個(gè)消息隊(duì)列topic優(yōu)先級屬性，然后按照高優(yōu)先級的消息隊(duì)列topic優(yōu)先消費(fèi)來將處理訂單的順序重新排列，達(dá)到高優(yōu)先級的訂單優(yōu)先處理的目的，而且保證了消息隊(duì)列還是分布式。同時(shí)，通過Json微服務(wù)調(diào)用工單拆分模塊，將優(yōu)先級高的訂單分配到優(yōu)先級高的消息隊(duì)列中。最后高優(yōu)先級的訂單會(huì)優(yōu)先執(zhí)行工單激活模塊，從而使多主機(jī)多節(jié)點(diǎn)同時(shí)工作，起到負(fù)載均衡和自動(dòng)切換的作用，實(shí)現(xiàn)高可用。

2 總體技術(shù)方案

通過Json微服務(wù)調(diào)用工單拆分接口內(nèi)的方法，工單拆分模塊接收到Json調(diào)用后會(huì)接收到該訂單要辦理的業(yè)務(wù)的相關(guān)數(shù)據(jù)信息，在這個(gè)模塊中有三部分要操作，即工單校驗(yàn)、工單拆分、工單入庫。接到上游系統(tǒng)發(fā)來的業(yè)務(wù)Json后，對輸入?yún)?shù)進(jìn)行校驗(yàn)，如果缺少參數(shù)，則返回；如果參數(shù)完整正確，則繼續(xù)。對工單進(jìn)行拆分，遍歷工單的服務(wù)編碼，并結(jié)合參數(shù)編碼匹配對應(yīng)的子服務(wù)，如果無，則直接將服務(wù)寫入服務(wù)列表；如果有，則生成新的子工單，并接入對應(yīng)的平臺(tái)和服務(wù)。遍歷工單服務(wù)過程中，拆分新的服務(wù)編碼時(shí)，如果發(fā)現(xiàn)已有工單存在對應(yīng)的平臺(tái)，則嘗試合并工單服務(wù)列表；如果已有工單服務(wù)中存在串行服務(wù)，則生成新的工單。遍歷完工單的服務(wù)編碼后，將業(yè)務(wù)號(hào)碼發(fā)送kafka中，kafka的topic為業(yè)務(wù)Json中指定的優(yōu)先級topic。

工單調(diào)度模塊采用多線程處理，工單獲取Get線程，工單完工Finish線程，以及網(wǎng)元redis隊(duì)列：in隊(duì)列和out隊(duì)列。Get線程根據(jù)號(hào)碼處理相關(guān)的業(yè)務(wù)，號(hào)碼來源是kafka消息，按照優(yōu)先級從高到低的順序依次消費(fèi)kafka中的存放業(yè)務(wù)號(hào)碼的topic，并根據(jù)號(hào)碼取出數(shù)據(jù)庫中關(guān)聯(lián)的未處理工單。根據(jù)網(wǎng)元類型將工單分類打包，并放到待發(fā)送隊(duì)列in隊(duì)列。Finish線程獲取待完工隊(duì)列out中的單個(gè)工單，并根據(jù)當(dāng)前工單的狀態(tài)做不同的處理。工單激活模塊會(huì)提取工單調(diào)度模塊中保存到in隊(duì)列中的工單數(shù)據(jù)進(jìn)行工單翻譯和工單交互。工單翻譯為將工單的服務(wù)翻譯為具體的指令集，工單交互為將工單的具體指令集發(fā)送到網(wǎng)元。

工單激活模塊收到待發(fā)送工單列表集合后，會(huì)檢查工單的服務(wù)名和當(dāng)前激活模塊服務(wù)名是否一致。對工單集合依次進(jìn)行翻譯，遍歷工單列表和遍歷工單的服務(wù)列表，完成工單服務(wù)編碼對應(yīng)的指令的映射、工單參數(shù)的變種和替換。遍歷工單的服務(wù)列表，對工單的每個(gè)服務(wù)編碼的多個(gè)指令進(jìn)行交互，并根據(jù)交互的結(jié)構(gòu)進(jìn)行綜合判斷。如果交互成功，則進(jìn)行下一個(gè)指令的交互；如果交互失敗，則轉(zhuǎn)換失敗原因和進(jìn)行容錯(cuò)判斷。所有工單的所有服務(wù)編碼對應(yīng)的所有指令交互完成后，進(jìn)行整個(gè)工單列表集合的返回。

2.1 消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用

聯(lián)指云化系統(tǒng)是一種基于消息優(yōu)先級隊(duì)列分布式交換網(wǎng)元調(diào)用技術(shù)的系統(tǒng)（如圖1所示），主要用到三份模塊化的代碼，即工單拆分、工單調(diào)度、工單激活這三個(gè)模塊，和一個(gè)數(shù)據(jù)緩存集群、一個(gè)消息隊(duì)列集群、數(shù)據(jù)持久化集群。本系統(tǒng)根據(jù)工具對業(yè)務(wù)的契合度和工具的優(yōu)缺點(diǎn)，決定選用spring-cloud來模塊化代碼，redis工具來作為數(shù)據(jù)緩存集群，kafka作為分布式消息隊(duì)列集群。通過Json微服務(wù)調(diào)用聯(lián)指云化系統(tǒng)，數(shù)據(jù)經(jīng)過該系統(tǒng)的處理后就會(huì)按照高優(yōu)先級到低優(yōu)先級的順序發(fā)送指令去外圍網(wǎng)元系統(tǒng)集合。

圖1 基于消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用的流程示意圖

2.2 消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用中工單拆分

運(yùn)行的spring-cloud程序接收營業(yè)接口表的Json微服務(wù)調(diào)用。如圖2，工單拆分模塊首先會(huì)對Json數(shù)據(jù)的完整性進(jìn)行校驗(yàn)，如參數(shù)的個(gè)數(shù)是否符合，如果參數(shù)為空，就直接返回，以及在日志中記錄該條數(shù)據(jù)的報(bào)錯(cuò)和流水號(hào)，以便后期的報(bào)錯(cuò)定位。如果參數(shù)不為空則代表接收到的數(shù)據(jù)是完整的，那么程序會(huì)繼續(xù)往下走。工單拆分的第二步是對接收到的工單數(shù)據(jù)進(jìn)行服務(wù)編碼匹配，工單數(shù)據(jù)有兩種類型，一種是服務(wù)編碼所代表的服務(wù)還能被細(xì)分成多個(gè)服務(wù)的，含有這種服務(wù)編碼的數(shù)據(jù)會(huì)在這一步操作中被工單拆分模塊拆分成多條數(shù)據(jù)。而另外一種數(shù)據(jù)則是服務(wù)編碼僅僅代表一個(gè)服務(wù)，無法再細(xì)分下去了，那么這條數(shù)據(jù)就會(huì)跳過這一步服務(wù)代碼拆分操作，直接進(jìn)入下一步操作。在工單拆分的第三步操作中，處理完的數(shù)據(jù)分為兩種類型的數(shù)據(jù)：一類是數(shù)據(jù)字段中網(wǎng)元字段相同的數(shù)據(jù)；另一類是網(wǎng)元字段不相同的數(shù)據(jù)。如果網(wǎng)元字段相同，那么代表這些數(shù)據(jù)將會(huì)發(fā)送去同一個(gè)目的地，同一個(gè)外圍網(wǎng)元，所以將這類數(shù)據(jù)進(jìn)行合并然后保存在數(shù)據(jù)庫中；如果網(wǎng)元字段不相同，則代表發(fā)往不同外圍網(wǎng)元，不進(jìn)行數(shù)據(jù)合并直接保存到數(shù)據(jù)庫中。在這一步進(jìn)行工單合并的用意是減少同一網(wǎng)元的數(shù)據(jù)多次發(fā)送，降低與外圍網(wǎng)元系統(tǒng)集合交互的次數(shù)，避免因?yàn)榫W(wǎng)絡(luò)原因而導(dǎo)致的不必要損失。工單拆分的最后一步操作是按照J(rèn)son中優(yōu)先級字段，將數(shù)據(jù)寫入到kafka的不同topic中（topic01，topic02，topic03，topic0n），優(yōu)先級依次遞減，到此工單拆分模塊執(zhí)行結(jié)束。

2.3 消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用中工單調(diào)度

當(dāng)數(shù)據(jù)經(jīng)過工單拆分模塊后會(huì)保存在兩個(gè)地方（如圖3所示）。

圖3 基于消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用中工單調(diào)度的流程示意圖

在工單調(diào)度模塊有2個(gè)線程，分別為Get線程與Fin線程，而保存在kafka中的數(shù)據(jù)就會(huì)觸發(fā)工單調(diào)度的Get線程。Get線程的第一步操作是按照優(yōu)先級進(jìn)行kafka消費(fèi)，采用算法讀取（topic01，topic02，topic03，topic0n），以此類推，Get線程沒有消費(fèi)到數(shù)據(jù)時(shí)會(huì)一直消費(fèi)kafka，當(dāng)消費(fèi)到了數(shù)據(jù)時(shí)就會(huì)進(jìn)行下一步操作，檢查該條數(shù)據(jù)是否有其他進(jìn)程在處理，如果有的話就跳過這條數(shù)據(jù)，處理下一條數(shù)據(jù)，如果沒有線程在處理這條數(shù)據(jù)的話就會(huì)根據(jù)這條數(shù)據(jù)中的號(hào)碼字段來撈取數(shù)據(jù)庫中未處理的工單，當(dāng)在數(shù)據(jù)庫中撈取不到該號(hào)碼的相關(guān)工單就跳過這條數(shù)據(jù)，處理下一條數(shù)據(jù)，如果在數(shù)據(jù)庫中撈取到該號(hào)碼的相關(guān)工單就會(huì)進(jìn)行下一步的判斷，檢查號(hào)碼是否有未完成的關(guān)鍵工單，如果有就跳過，處理下一條數(shù)據(jù)，如果沒有就代表這條工單已經(jīng)全部完成，則更新工單在redis和數(shù)據(jù)庫中的狀態(tài)，并按網(wǎng)元分類打包，放到待發(fā)送隊(duì)列redis–in列表。

工單調(diào)度中的Fin線程會(huì)一直去獲取redis-out列表中的工單，第一步操作會(huì)判斷這條數(shù)據(jù)是否為關(guān)鍵工單，當(dāng)不是關(guān)鍵工單的時(shí)候會(huì)跳過，繼續(xù)獲取下一條數(shù)據(jù)，當(dāng)獲取到關(guān)鍵工單數(shù)據(jù)的時(shí)候會(huì)進(jìn)行第二步操作，判斷工單狀態(tài)（錯(cuò)誤，成功，超時(shí)等）。如果這條數(shù)據(jù)工單狀態(tài)為錯(cuò)誤，就回單，也就是更新數(shù)據(jù)庫中的狀態(tài)為錯(cuò)誤；如果這條數(shù)據(jù)工單狀態(tài)為正確，會(huì)進(jìn)行下一步判斷，是否為其他工單的前置工單，是的話就更新后置工單到redis-in列表中然后回到Fin線程最開始，不是話就檢查所有工單狀態(tài)，符合回單就更新表狀態(tài)，不符合就刪除該號(hào)碼關(guān)聯(lián)定單中當(dāng)前定單；如果這條數(shù)據(jù)的工單狀態(tài)為超時(shí)則直接刪除該號(hào)碼關(guān)聯(lián)定單中當(dāng)前定單，經(jīng)過第二步數(shù)據(jù)狀態(tài)判斷的操作后就會(huì)進(jìn)行第三步操作，檢查該號(hào)碼關(guān)聯(lián)的訂單狀態(tài)，如果還有沒完成的關(guān)鍵訂單就繼續(xù)等待，如果關(guān)鍵訂單均完成后就會(huì)檢查是否有同一個(gè)號(hào)碼的其他訂單，有的話就進(jìn)入號(hào)碼二次處理隊(duì)列，沒有的話就刪除同號(hào)標(biāo)志。

2.4 消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用中工單激活

如圖4所示，在工單拆分模塊中處理完的數(shù)據(jù)，也就是圖中接收到的Json請求并且通過了服務(wù)名校驗(yàn)的數(shù)據(jù)，這些數(shù)據(jù)存到了redis-in 列表中，工單激活模塊會(huì)一直去redis-in 列表中獲取數(shù)據(jù)，當(dāng)獲取到一條數(shù)據(jù)的時(shí)候就會(huì)進(jìn)行第一步操作——工單翻譯，首先會(huì)處理數(shù)據(jù)中的特殊服務(wù)，如果在工單服務(wù)列表中匹配不到的話就跳過，如果匹配到了就判斷特殊服務(wù)類型并處理服務(wù)列表；其次是處理擴(kuò)展參數(shù)，如果遍歷工單參數(shù)列表后發(fā)現(xiàn)變種參數(shù)的話就直接生成變種參數(shù)然后加入擴(kuò)展參數(shù)列表，如果沒發(fā)現(xiàn)變種參數(shù)就直接加入擴(kuò)展參數(shù)列表；再次就是獲取服務(wù)編碼的配置，替換服務(wù)對象；最后是指令代碼參數(shù)替換，擴(kuò)展參數(shù)替換指令模板。工單翻譯運(yùn)行到這里就完成了，每條工單被翻譯成能夠被網(wǎng)元理解的指令。

圖4 基于消息優(yōu)先級隊(duì)列的分布式交換網(wǎng)元調(diào)用中工單激活的流程示意圖

當(dāng)工單翻譯完了之后就會(huì)和對應(yīng)的網(wǎng)元進(jìn)行工單交互，首先會(huì)去初始化登錄的配置，獲取線程所屬socket和網(wǎng)元的配置，然后就能夠登錄到對應(yīng)的網(wǎng)元，并根據(jù)交互的結(jié)構(gòu)進(jìn)行綜合判斷，如果交互成功，則進(jìn)行下一個(gè)指令的交互。如果交互失敗，則轉(zhuǎn)換失敗原因和進(jìn)行容錯(cuò)判斷。所有工單的所有服務(wù)編碼對應(yīng)的所有指令交互完成后，進(jìn)行整個(gè)工單列表集合的返回。

2.5 運(yùn)行效果

高峰期工單處理能力比未改造消息優(yōu)先級隊(duì)列時(shí)：原來100萬/天，提升到＞1000萬/天（10倍）；高峰期工單處理時(shí)長：原來14 秒，縮短到＜2 秒（7 倍）；高峰期聯(lián)指處理能力：原來100 萬/天，提升到＞1000 萬/天（10 倍）；高峰期指令處理時(shí)長（含網(wǎng)元時(shí)間）：原來17 秒，縮短到＜3 秒（6 倍）；工單并發(fā)處理數(shù)：原來10TPS，提升到＞200TPS（20倍）。

3 結(jié)語

本系統(tǒng)通過Json微服務(wù)調(diào)用工單拆分模塊，將優(yōu)先級高的訂單分配到優(yōu)先級高的消息隊(duì)列中，然后工單調(diào)度系統(tǒng)按照高優(yōu)先級隊(duì)列優(yōu)先處理原則進(jìn)行工單的調(diào)度，最后高優(yōu)先級的訂單會(huì)優(yōu)先執(zhí)行工單激活模塊，這樣既保證了消息的隊(duì)列分布式處理，又能將訂單進(jìn)行排序，以提高系統(tǒng)效率，強(qiáng)化系統(tǒng)能力，擴(kuò)大系統(tǒng)使用范圍，同時(shí)具備云平臺(tái)的高效率、高可用、高擴(kuò)展特點(diǎn)，簡化運(yùn)維操作，釋放人力資源。高效率即是模塊微服務(wù)化，輕Json 格式調(diào)用，去除耗時(shí)的數(shù)據(jù)庫掃表操作，使用中間件緩存。高可用即是系統(tǒng)分布式部署，多主機(jī)多節(jié)點(diǎn)同時(shí)工作，起到負(fù)載均衡和自動(dòng)切換的作用，實(shí)現(xiàn)高可用。