基于消息優(yōu)先級隊列分布式交換網(wǎng)元調(diào)用系統(tǒng)的研究

胡佳

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

1 引言

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

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

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

2 總體技術(shù)方案

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

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

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

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

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

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

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

運行的spring-cloud程序接收營業(yè)接口表的Json微服務(wù)調(diào)用。如圖2，工單拆分模塊首先會對Json數(shù)據(jù)的完整性進行校驗，如參數(shù)的個數(shù)是否符合，如果參數(shù)為空，就直接返回，以及在日志中記錄該條數(shù)據(jù)的報錯和流水號，以便后期的報錯定位。如果參數(shù)不為空則代表接收到的數(shù)據(jù)是完整的，那么程序會繼續(xù)往下走。工單拆分的第二步是對接收到的工單數(shù)據(jù)進行服務(wù)編碼匹配，工單數(shù)據(jù)有兩種類型，一種是服務(wù)編碼所代表的服務(wù)還能被細(xì)分成多個服務(wù)的，含有這種服務(wù)編碼的數(shù)據(jù)會在這一步操作中被工單拆分模塊拆分成多條數(shù)據(jù)。而另外一種數(shù)據(jù)則是服務(wù)編碼僅僅代表一個服務(wù)，無法再細(xì)分下去了，那么這條數(shù)據(jù)就會跳過這一步服務(wù)代碼拆分操作，直接進入下一步操作。在工單拆分的第三步操作中，處理完的數(shù)據(jù)分為兩種類型的數(shù)據(jù)：一類是數(shù)據(jù)字段中網(wǎng)元字段相同的數(shù)據(jù)；另一類是網(wǎng)元字段不相同的數(shù)據(jù)。如果網(wǎng)元字段相同，那么代表這些數(shù)據(jù)將會發(fā)送去同一個目的地，同一個外圍網(wǎng)元，所以將這類數(shù)據(jù)進行合并然后保存在數(shù)據(jù)庫中；如果網(wǎng)元字段不相同，則代表發(fā)往不同外圍網(wǎng)元，不進行數(shù)據(jù)合并直接保存到數(shù)據(jù)庫中。在這一步進行工單合并的用意是減少同一網(wǎng)元的數(shù)據(jù)多次發(fā)送，降低與外圍網(wǎng)元系統(tǒng)集合交互的次數(shù)，避免因為網(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)先級隊列的分布式交換網(wǎng)元調(diào)用中工單調(diào)度

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

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

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

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

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

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

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

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

2.5 運行效果

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

3 結(jié)語

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