劉 瀟

(江蘇省疾病預(yù)防控制中心 公共衛(wèi)生信息所, 南京 210009)

隨著疾控信息化工作的不斷深入, 疾控的傳染病、公共衛(wèi)生突發(fā)事件、慢病、計(jì)劃免疫以及精神衛(wèi)生等業(yè)務(wù)條線的信息系統(tǒng)在不斷地建立與完善, 疾控信息化標(biāo)準(zhǔn)體系[1,2]的建立與完善有力地推動(dòng)了全民健康信息化中公共衛(wèi)生的數(shù)據(jù)整合. 在當(dāng)前各行業(yè)協(xié)作日益緊密、各級(jí)疾控一體化集成日漸成熟的大背景下, 疾控中心各類(lèi)的數(shù)據(jù)共享與交換[3,4]需求也隨之而來(lái). 根據(jù)不同的業(yè)務(wù)需求, 各個(gè)信息系統(tǒng)需要調(diào)用不同來(lái)源的接口來(lái)完成數(shù)據(jù)的下載、上傳或核驗(yàn)等操作.在數(shù)據(jù)量比較小、任務(wù)實(shí)時(shí)性要求比較低的情況下,全量數(shù)據(jù)逐條調(diào)用數(shù)據(jù)接口并記錄接口反饋信息的模式可以滿足業(yè)務(wù)需求, 但是當(dāng)數(shù)據(jù)量比較大并且任務(wù)實(shí)時(shí)性要求比較高的情況下, 比如: 疫情期間, 全省億級(jí)數(shù)量的常住人口庫(kù)的全量數(shù)據(jù)需要周期性調(diào)用通信管理接口或核酸檢測(cè)查詢接口以獲得個(gè)人行程記錄與核酸檢測(cè)的相關(guān)信息, 或是在特定的時(shí)間內(nèi), 某個(gè)月增百萬(wàn)級(jí)隨訪數(shù)據(jù)的業(yè)務(wù)系統(tǒng)的大量的隨訪信息需要全部上傳至指定的平臺(tái), 逐條調(diào)用數(shù)據(jù)接口的模式效率太低, 無(wú)法在規(guī)定的時(shí)間內(nèi)完成任務(wù), 如何利用有限的硬件資源高效地完成數(shù)據(jù)交換任務(wù)成為了疾控在信息化建設(shè)中面臨的一個(gè)問(wèn)題.

在有限的硬件資源下, 解決這個(gè)問(wèn)題的思路是讓數(shù)據(jù)交換任務(wù)并發(fā)執(zhí)行, 直接在服務(wù)器上為每一個(gè)數(shù)據(jù)交換任務(wù)分配一個(gè)線程并同時(shí)啟動(dòng)大量線程去完成數(shù)據(jù)交換的方法會(huì)導(dǎo)致服務(wù)器壓力過(guò)大, 線程的運(yùn)行缺乏有效的控制, 線程的創(chuàng)建與銷(xiāo)毀都會(huì)造成系統(tǒng)開(kāi)銷(xiāo), 操作系統(tǒng)對(duì)大量線程的頻繁的切換與調(diào)度會(huì)給CPU 帶來(lái)沉重的負(fù)擔(dān), 容易造成服務(wù)卡頓或服務(wù)器宕機(jī). 本文基于線程池與消息中間件技術(shù)建立一個(gè)數(shù)據(jù)交換的并發(fā)處理模型, 使用Java 線程池去控制數(shù)據(jù)交換任務(wù)的并發(fā)處理, 并引用消息中間件Kafka 作為中間件來(lái)記錄數(shù)據(jù)交換結(jié)果, 進(jìn)一步提高任務(wù)完成的效率, 通過(guò)實(shí)驗(yàn)的對(duì)比證明該模型的可行性與高效性.

1 技術(shù)簡(jiǎn)介

1.1 線程池介紹

線程池技術(shù)是一種設(shè)計(jì)程序并發(fā)運(yùn)行的技術(shù), 其核心思想是對(duì)已有線程的復(fù)用來(lái)避免大量線程創(chuàng)建與銷(xiāo)毀帶來(lái)的系統(tǒng)開(kāi)銷(xiāo), 在CPU 上創(chuàng)建和結(jié)束線程造成的開(kāi)銷(xiāo)是創(chuàng)建或銷(xiāo)毀任務(wù)的18 至100 倍[5], 而且通過(guò)任務(wù)進(jìn)行同步的開(kāi)銷(xiāo)也遠(yuǎn)低于同步多個(gè)線程的開(kāi)銷(xiāo), 因此線程池技術(shù)能夠更好地支持細(xì)粒度的任務(wù)并發(fā)[6]. 常見(jiàn)的線程池一般主要包括4 個(gè)部分: 線程管理器、工作線程、任務(wù)接口和輸入輸出任務(wù)隊(duì)列, 在啟動(dòng)時(shí)線程池創(chuàng)建若干數(shù)量的空閑線程, 當(dāng)任務(wù)到達(dá)時(shí)利用已經(jīng)創(chuàng)建的線程執(zhí)行任務(wù), 任務(wù)處理完成后, 該線程會(huì)被線程池回收用來(lái)執(zhí)行下一個(gè)任務(wù)以達(dá)到線程復(fù)用的效果, 同時(shí)線程池還要對(duì)任務(wù)隊(duì)列的大小、空閑線程的銷(xiāo)毀、新線程的創(chuàng)建以及對(duì)任務(wù)的拒絕策略等進(jìn)行管理.

Java 從JDK 1.5 版本開(kāi)始在java.util.concurrent 包中提供了對(duì)線程池功能的支持[7], 相關(guān)類(lèi)的繼承關(guān)系如圖1 所示, 其中ThreadPoolExecutor 是最核心的一個(gè)類(lèi), Java 通過(guò)封裝ThreadPoolExecutor 類(lèi)提供了SingleThreadExecutor、CachedThreadPool、Fixed ThreadPool 以及ScheduledThreadPool 這4 類(lèi)適合特定場(chǎng)景的線程池供編程人員調(diào)用, 同時(shí)Java 也支持編程人員重寫(xiě)ThreadPoolExecutor 的構(gòu)造方法, 通過(guò)設(shè)置構(gòu)造參數(shù)自定義線程池.

圖1 Java 線程池UML 靜態(tài)類(lèi)圖

ThreadPoolExecutor 類(lèi)構(gòu)造方法的主要的構(gòu)造參數(shù)如下:

corePoolSize: 核心線程數(shù), 即常駐線程池的工作線程數(shù)量.

maximumPoolSize: 最大線程數(shù), 即某一時(shí)刻, 當(dāng)任務(wù)大于線程池當(dāng)前存在的工作線程數(shù)時(shí), 線程池中的工作線程可以增加到的最大值.

keepAliveTime: 當(dāng)線程數(shù)大于核心線程數(shù)時(shí), 空閑的工作線程等待新任務(wù)的最長(zhǎng)時(shí)間, 超過(guò)這個(gè)時(shí)間空閑線程沒(méi)有接到任務(wù)就會(huì)被銷(xiāo)毀, 線程池只保留核心線程數(shù)的工作線程數(shù)量.

workQueue: 任務(wù)隊(duì)列, 即線程池中的工作線程的數(shù)量已經(jīng)達(dá)到最大線程數(shù)時(shí), 任務(wù)的等待隊(duì)列.

threadFactory: 線程工廠, 可以用來(lái)自定義線程池中線程的命名方式, 優(yōu)先級(jí)等屬性.

Handler: 拒絕策略, 即線程池中的工作線程的數(shù)量已經(jīng)達(dá)到最大線程數(shù)且任務(wù)隊(duì)列已滿的情況下, 線程池對(duì)超出線程池處理能力的任務(wù)所做的處理策略.

1.2 消息中間件介紹

消息中間件是可以在不同系統(tǒng)之間進(jìn)行消息傳遞的一類(lèi)組件, 它利用高效、可靠的消息傳遞機(jī)制進(jìn)行平臺(tái)無(wú)關(guān)的數(shù)據(jù)交流[8], 消息生產(chǎn)者定向發(fā)送數(shù)據(jù), 消息消費(fèi)者獲取并消費(fèi)數(shù)據(jù), 基于數(shù)據(jù)通信進(jìn)行分布式系統(tǒng)的集成. 消息中間件的消息傳遞主要有兩種模式,分別是點(diǎn)對(duì)點(diǎn)模式和發(fā)布-訂閱模式. 目前比較主流的分布式消息中間件有Kafka, RabbitMQ, ActiveMQ 等.

Kafka 是一個(gè)分布式的消息發(fā)布-訂閱模式[9]的中間件系統(tǒng). Kafka 在主題中保存消息的信息, 生產(chǎn)者向主題寫(xiě)入數(shù)據(jù), 消費(fèi)者從主題讀取數(shù)據(jù), 從而實(shí)現(xiàn)數(shù)據(jù)傳輸.

高性能、高吞吐、低延時(shí)是Kafka 的顯著的特性,雖然Kafka 的消息保存在磁盤(pán)上, 但是由于采用了順序?qū)懭搿MFiles (memory mapped files)、Zero Copy、批量壓縮等技術(shù)優(yōu)化了讀寫(xiě)性能[10], 使其可以突破傳統(tǒng)的數(shù)據(jù)庫(kù)、消息隊(duì)列等數(shù)據(jù)引擎所受限的磁盤(pán)IO瓶頸, 即使是部署在普通的單機(jī)服務(wù)器上, Kafka 也能輕松支持每秒百萬(wàn)級(jí)的寫(xiě)入請(qǐng)求[11], 讀寫(xiě)速度超過(guò)大部分的消息中間件, 這種特性使得Kafka 在海量數(shù)據(jù)場(chǎng)景中應(yīng)用廣泛.

2 模型設(shè)計(jì)與實(shí)現(xiàn)

2.1 模型設(shè)計(jì)

疾控信息化工作中處理數(shù)據(jù)交換的基本流程是:從數(shù)據(jù)庫(kù)中分批取出需要調(diào)用數(shù)據(jù)接口的數(shù)據(jù), 為批次中的每一條數(shù)據(jù)創(chuàng)建一個(gè)數(shù)據(jù)交換任務(wù), 任務(wù)主要包括調(diào)用接口獲得反饋信息、將反饋信息回寫(xiě)數(shù)據(jù)庫(kù)進(jìn)行持久化兩個(gè)步驟.

由于各數(shù)據(jù)交換任務(wù)相互之間的無(wú)關(guān)性, 可以在調(diào)用的數(shù)據(jù)接口可承載的并發(fā)調(diào)用范圍內(nèi), 使數(shù)據(jù)交換任務(wù)并發(fā)進(jìn)行以提高效率, 并在數(shù)據(jù)交換任務(wù)的反饋信息持久化階段將反饋信息寫(xiě)入吞吐量更高的消息中間件進(jìn)行存儲(chǔ), 進(jìn)一步縮短數(shù)據(jù)交換任務(wù)的運(yùn)行時(shí)間以提高效率.

在圖2 中, 通過(guò)一個(gè)數(shù)據(jù)交換調(diào)度控制程序建立并初始化數(shù)據(jù)交換任務(wù)的線程池, 在進(jìn)行數(shù)據(jù)交換任務(wù)時(shí), 為從數(shù)據(jù)庫(kù)取出的批量數(shù)據(jù)構(gòu)造數(shù)據(jù)交換任務(wù),并將任務(wù)交給線程池進(jìn)行并發(fā)處理的調(diào)度, 數(shù)據(jù)接口的反饋信息寫(xiě)入中間件進(jìn)行保存, 不同的數(shù)據(jù)消費(fèi)者進(jìn)程可以異步消費(fèi)消息中間以獲取反饋信息, 按照不同的業(yè)務(wù)需求進(jìn)行日志信息持久化到數(shù)據(jù)庫(kù)或者實(shí)時(shí)進(jìn)行交換日志的統(tǒng)計(jì)與分析等操作.

圖2 數(shù)據(jù)交換并發(fā)處理模型

2.2 技術(shù)實(shí)現(xiàn)

數(shù)據(jù)交換調(diào)度控制程序用Java 設(shè)計(jì), 使用Java 線程池與Kafka 對(duì)模型進(jìn)行實(shí)現(xiàn), 模型實(shí)現(xiàn)主要包含數(shù)據(jù)交換任務(wù)構(gòu)造、Kafka 調(diào)用以及數(shù)據(jù)交換線程池3 個(gè)部分.

2.2.1 數(shù)據(jù)交換任務(wù)構(gòu)造

封裝數(shù)據(jù)交換任務(wù)的類(lèi)需要實(shí)現(xiàn)Runnable 接口以保證其可以在實(shí)例化后被線程池工作線程所調(diào)用,在該類(lèi)的構(gòu)造器中傳遞具體的Kafka 連接以及數(shù)據(jù)接口調(diào)用所需要的參數(shù), 并實(shí)現(xiàn)Runnable 接口的run 方法完成具體數(shù)據(jù)接口調(diào)用與反饋信息的記錄, 其核心代碼如下:

?public class DSTask implements Runnable{ //數(shù)據(jù)交換任務(wù)封裝類(lèi)public DSTask(KafkaProducer producer, String id, ....){ //通過(guò)構(gòu)造器為實(shí)例的屬性賦值this.producer=producer; //Kafka 連接對(duì)象this.id=id; //數(shù)據(jù)在數(shù)據(jù)庫(kù)的主鍵號(hào).... //數(shù)據(jù)接口調(diào)用的各項(xiàng)參數(shù)}@Override public void run() { //實(shí)現(xiàn)Runnable 接口的run 方法sendData(); //該方法獲取構(gòu)造器傳遞的屬性完成數(shù)據(jù)交換任務(wù)}

2.2.2 Kafka 調(diào)用

在數(shù)據(jù)交換任務(wù)封裝類(lèi)的sendData 方法中調(diào)用Kafka api 提供的send 方法記錄反饋信息, String 類(lèi)型topicName 為Kafka 的相關(guān)主題名, String 類(lèi)型context 為數(shù)據(jù)交換任務(wù)最終按約定格式拼接好的反饋信息, 其核心代碼如下:

2.2.3 數(shù)據(jù)交換線程池

通過(guò)參數(shù)設(shè)置自定義ThreadPoolExecutor 類(lèi)實(shí)例化線程池來(lái)控制數(shù)據(jù)交換任務(wù)并發(fā)處理. 由于數(shù)據(jù)交換任務(wù)需要連續(xù)穩(wěn)定的處理, 線程池的核心線程數(shù)和最大線程數(shù)設(shè)為相同值, 即線程池中的常駐的工作線程數(shù), 這個(gè)值的大小在運(yùn)行前需要由用戶綜合考慮所調(diào)用數(shù)據(jù)接口能承載的并發(fā)訪問(wèn)量, 以及當(dāng)前任務(wù)所運(yùn)行的服務(wù)器的CPU 核數(shù)來(lái)設(shè)定, 在數(shù)據(jù)接口并發(fā)訪問(wèn)的承載范圍內(nèi), 在實(shí)際工程應(yīng)用中一般遵循如式(1)所示[12]:

線程池的任務(wù)隊(duì)列的大小設(shè)置為每批要調(diào)用數(shù)據(jù)接口的數(shù)據(jù)的數(shù)量, 以保證所有的數(shù)據(jù)交換任務(wù)都會(huì)被任務(wù)隊(duì)列容納, 等待線程池的有效調(diào)度, 這樣可以直接使用線程池默認(rèn)的拒絕策略, 不需要再設(shè)計(jì)拒絕策略去處理線程池?zé)o法處理的數(shù)據(jù)交換任務(wù).

線程池核心代碼如下:

?ThreadPoolExecutor executor = new ThreadPoolExecutor(threadNum,threadNum, 10, TimeUnit.SECONDS,new LinkedBlockingQueue(queueCapacity),Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy()); //實(shí)例化線程池, int 類(lèi)型threadNum 為工作線程數(shù), int 類(lèi)型queueCapacity 為隊(duì)列容量, 使用默認(rèn)的拒絕策略DiscardPolicy while(rs.next()){ //遍歷從數(shù)據(jù)庫(kù)取出的一批次的數(shù)據(jù)String id = rs.getString("id"); //獲取相關(guān)參數(shù)....executor.execute(new DSTask(producer, id, ....)); //實(shí)例化數(shù)據(jù)交換任務(wù)并交給線程池調(diào)度執(zhí)行}

3 仿真實(shí)驗(yàn)

為測(cè)試該模型處理數(shù)據(jù)交換任務(wù)的效率, 在疾控內(nèi)部局域網(wǎng)部署應(yīng)用進(jìn)行測(cè)試, 應(yīng)用部署的服務(wù)器配置: 4 核CPU, 內(nèi)存8 GB, 操作系統(tǒng): 64 位Linux CentOS 7.7, JDK 版本: Openjdk 1.8, 測(cè)試從疾控內(nèi)網(wǎng)某業(yè)務(wù)庫(kù)(業(yè)務(wù)庫(kù)版本: MySQL 8.0.18)批量取出5 000 條個(gè)人信息數(shù)據(jù)調(diào)用在公網(wǎng)發(fā)布的疫苗接種記錄查詢接口獲取個(gè)人某疫苗首次接種記錄的相關(guān)信息, 在逐條處理以及使用線程池模型進(jìn)行處理、接口反饋的結(jié)果回寫(xiě)數(shù)據(jù)庫(kù)或?qū)懭隟afka 等一些不同的情況下, 分別進(jìn)行如下仿真實(shí)驗(yàn):

實(shí)驗(yàn)1. 數(shù)據(jù)接口反饋信息回寫(xiě)數(shù)據(jù)庫(kù), 單線程逐條處理以及使用線程池在工作線程數(shù)取不同值的情況下的運(yùn)行時(shí)間對(duì)比, 運(yùn)行時(shí)間皆為5 次實(shí)驗(yàn)的平均值,數(shù)據(jù)如表1 所示.

表1 不同工作線程數(shù)運(yùn)行時(shí)間對(duì)比

很顯然, 線程池處理完成數(shù)據(jù)交換任務(wù)的效率明顯優(yōu)于單線程逐條處理, 且在實(shí)際接口的實(shí)際條件以及4 核CPU 的硬件資源條件下, 在工作線程數(shù)設(shè)為4 時(shí)的運(yùn)行效率已達(dá)到最佳.

實(shí)驗(yàn)2. 在線程池在工作線程數(shù)取最佳值的情況下, 數(shù)據(jù)接口反饋信息回寫(xiě)數(shù)據(jù)庫(kù)與寫(xiě)入Kafka (版本:Kafka 2.5.0)的運(yùn)行時(shí)間對(duì)比, 運(yùn)行時(shí)間皆為5 次實(shí)驗(yàn)的平均值, 數(shù)據(jù)如表2 所示.

表2 反饋信息回寫(xiě)數(shù)據(jù)庫(kù)與寫(xiě)入Kafka 運(yùn)行時(shí)間對(duì)比

對(duì)比兩者的運(yùn)行時(shí)間可以看出, 將數(shù)據(jù)接口反饋信息寫(xiě)入Kafka 可以極大地提高了數(shù)據(jù)交換任務(wù)完成的效率.

4 模型應(yīng)用

在疾控的數(shù)據(jù)交換工作中對(duì)模型進(jìn)行實(shí)際應(yīng)用時(shí),工程師根據(jù)需要進(jìn)行數(shù)據(jù)交換任務(wù)的數(shù)據(jù)總量, 綜合考慮部署數(shù)據(jù)交換應(yīng)用程序的服務(wù)器內(nèi)存情況, 對(duì)數(shù)據(jù)進(jìn)行批次的劃分, 確定每一批完成數(shù)據(jù)交換任務(wù)的數(shù)量與線程池任務(wù)隊(duì)列的容量, 并根據(jù)服務(wù)器CPU 的核數(shù)與需要調(diào)用的數(shù)據(jù)接口的實(shí)測(cè)情況確定線程池工作線程的數(shù)量, 設(shè)計(jì)數(shù)據(jù)交換調(diào)度控制程序. 如圖3 所示, 數(shù)據(jù)交換調(diào)度控制程序在初始化各類(lèi)連接并建立線程池后, 按照預(yù)設(shè)的批次, 分批對(duì)數(shù)據(jù)進(jìn)行數(shù)據(jù)交換任務(wù)的處理, 為了判斷線程池是否已完成當(dāng)前批次的所有數(shù)據(jù)交換任務(wù), 可以設(shè)置一個(gè)線程安全的全局變量, 每次數(shù)據(jù)交換任務(wù)完成時(shí)對(duì)這個(gè)變量進(jìn)行累加操作, 數(shù)據(jù)交換調(diào)度控制程序通過(guò)讀取這個(gè)變量值來(lái)獲取線程池的當(dāng)前狀態(tài), 如果當(dāng)前批次的任務(wù)尚未全部完成, 調(diào)度控制程序執(zhí)行自旋等待操作, 等待當(dāng)前批次的任務(wù)全部完成, 線程池處于空閑狀態(tài)后, 獲取下一批次的數(shù)據(jù)繼續(xù)進(jìn)行, 直至所有批次的數(shù)據(jù)全部完成.

圖3 數(shù)據(jù)交換調(diào)度控制程序流程設(shè)計(jì)

圖4 展示的是實(shí)際工作中某重點(diǎn)人群庫(kù)數(shù)據(jù)使用該模型調(diào)用新冠疫苗接種查詢接口獲取個(gè)人新冠疫苗第一針接種結(jié)果在Kafka 相關(guān)主題中的存儲(chǔ)情況, 該項(xiàng)數(shù)據(jù)交換任務(wù)按約定的格式記錄了個(gè)人信息在業(yè)務(wù)庫(kù)的主鍵號(hào), 調(diào)用接口的匹配標(biāo)識(shí), 以及調(diào)用接口所獲取的接種新冠疫苗第一針的疫苗廠商、接種時(shí)間、接種單位等信息, 各數(shù)據(jù)項(xiàng)之間插入制表符以便在信息消費(fèi)時(shí)進(jìn)行解析.

圖4 Kafka 記錄的反饋信息展示

5 結(jié)束語(yǔ)

針對(duì)疾控中心在處理大規(guī)模數(shù)據(jù)交換時(shí)傳統(tǒng)的處理模式效率不高, 難以及時(shí)完成任務(wù)的問(wèn)題, 本文根據(jù)數(shù)據(jù)交換任務(wù)的特點(diǎn)設(shè)計(jì)了一個(gè)數(shù)據(jù)交換任務(wù)的并發(fā)處理模型, 并使用Java 線程池與消息中間件Kafka 給出了模型的具體實(shí)現(xiàn). 該模型已成功應(yīng)用在江蘇省疾控中心的數(shù)據(jù)交換的處理中, 實(shí)踐表明, 模型具有良好的數(shù)據(jù)交換任務(wù)并發(fā)控制與處理能力, 進(jìn)行數(shù)據(jù)交換的數(shù)據(jù)量越大, 其優(yōu)勢(shì)越明顯. 在不大幅度增加硬件成本的前提下, 該模型適用面廣, 可用于各類(lèi)型的數(shù)據(jù)換的處理與控制, 在保證服務(wù)穩(wěn)定性的同時(shí)可以有效地提高數(shù)據(jù)交換的處理能力.