基于Netty和Kafka的工程機(jī)械車聯(lián)網(wǎng)數(shù)據(jù)采集系統(tǒng)設(shè)計(jì)方案

王 飛，孫嬌嬌，丁文文

（1徐工漢云技術(shù)股份有限公司，江蘇 徐州 221004；2中國礦業(yè)大學(xué)經(jīng)濟(jì)管理學(xué)院，江蘇 徐州 221116；3淮北師范大學(xué)數(shù)學(xué)科學(xué)學(xué)院，安徽 淮北 235000）

0 引言

目前，物聯(lián)網(wǎng)已經(jīng)成為全球公認(rèn)的可以推動(dòng)經(jīng)濟(jì)發(fā)展的新引擎。有數(shù)據(jù)顯示，2021年，全球物聯(lián)網(wǎng)設(shè)備連接數(shù)量高達(dá)122億個(gè)，據(jù)GSMA（Global System for Mobile communications Association）預(yù)測(cè)，2025年全球物聯(lián)網(wǎng)設(shè)備（包括蜂窩及非蜂窩）聯(lián)網(wǎng)數(shù)量將達(dá)到約246億個(gè)。萬物互聯(lián)成為未來發(fā)展的重要方向。車聯(lián)網(wǎng)作為物聯(lián)網(wǎng)的一個(gè)分支，近幾年來取得了長(zhǎng)足的發(fā)展，也成為研究和應(yīng)用的熱點(diǎn)。數(shù)據(jù)采集作為車聯(lián)網(wǎng)的“第一關(guān)”至關(guān)重要，既要保證數(shù)據(jù)的準(zhǔn)確性、時(shí)效性，又要保證系統(tǒng)的穩(wěn)定性、可靠性。近年來，Netty作為一款高性能網(wǎng)絡(luò)應(yīng)用程序框架備受關(guān)注，其實(shí)現(xiàn)了一套高性能內(nèi)存管理機(jī)制，支持百萬級(jí)數(shù)據(jù)并發(fā)。如甄凱成等[1]利用Netty網(wǎng)絡(luò)通信庫構(gòu)造高性能的收集端網(wǎng)絡(luò)通信處理服務(wù)并設(shè)計(jì)出可供采集終端與收集端使用的通信協(xié)議，在萬級(jí)別連接情況下，能正常工作且保持較快的響應(yīng)速度；胡喜明等[2]采用基于響應(yīng)式編程的Reactor-netty組件作為RPC底層通信框架，設(shè)計(jì)并實(shí)現(xiàn)一種動(dòng)態(tài)的負(fù)載均衡方案，相較以Netty作為通信框架的RPC系統(tǒng)在性能方面有一定提升；Zhang[3]等通過使用Netty設(shè)計(jì)了一種可擴(kuò)展為虛擬實(shí)驗(yàn)和物聯(lián)實(shí)驗(yàn)的一體化系統(tǒng)。

工程機(jī)械車聯(lián)網(wǎng)平臺(tái)一般包含數(shù)據(jù)采集、數(shù)據(jù)解析、業(yè)務(wù)處理和數(shù)據(jù)展示等基本功能模塊，數(shù)據(jù)采集模塊作為系統(tǒng)中數(shù)據(jù)接入的第一“門檻”至關(guān)重要。傳統(tǒng)的接入方式一般在平臺(tái)側(cè)建立socket通訊服務(wù)，用于接收車載終端發(fā)送數(shù)據(jù)，但這種方式在對(duì)通訊鏈路的監(jiān)控和管理方面比較困難，尤其在處理高并發(fā)、高吞吐量的數(shù)據(jù)時(shí)，系統(tǒng)的性能更加難以保證。因此，多數(shù)系統(tǒng)采用Netty基礎(chǔ)通訊框架對(duì)通訊鏈路進(jìn)行管控，但是在面臨高并發(fā)、高吞吐量的場(chǎng)景下，系統(tǒng)資源使用如何達(dá)到最優(yōu)以及整個(gè)數(shù)據(jù)流程的設(shè)計(jì)合理性方面，一直是業(yè)內(nèi)研究人員探索的課題。本文設(shè)計(jì)了一種基于Netty技術(shù)的數(shù)據(jù)采集和處理系統(tǒng)，采用“負(fù)載均衡”對(duì)設(shè)備IP層的數(shù)據(jù)進(jìn)行分發(fā)[4]，較好應(yīng)對(duì)了高并發(fā)訪問和提高了資源利用率[5]。系統(tǒng)設(shè)計(jì)上充分發(fā)揮Netty技術(shù)低延遲、高吞吐、分層且可擴(kuò)展的特性。另外，系統(tǒng)采用Redis緩存策略，提高數(shù)據(jù)庫IO訪問性能，利用Kafka發(fā)布訂閱模式將系統(tǒng)與業(yè)務(wù)處理進(jìn)行解耦，并充分發(fā)揮Kafka消息隊(duì)列在Linux系統(tǒng)中零拷貝技術(shù)，提升了系統(tǒng)整體性能，從而實(shí)現(xiàn)工程機(jī)械行業(yè)車聯(lián)網(wǎng)數(shù)據(jù)的實(shí)時(shí)采集。

1 關(guān)鍵技術(shù)

1.1 Netty技術(shù)

Netty是Jboss提供的一個(gè)java開源框架，提供了異步的、事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用程序框架和工具，用以快速開發(fā)高性能、高可用性的網(wǎng)絡(luò)服務(wù)器和客戶端程序。Netty作為基于NIO的編程框架，使用Netty可以快速開發(fā)出一個(gè)網(wǎng)絡(luò)應(yīng)用，用戶可以通過Future-Listener機(jī)制或者通知機(jī)制獲得I/O操作結(jié)果[6]。

分布式大數(shù)據(jù)計(jì)算引擎是科研機(jī)構(gòu)、互聯(lián)網(wǎng)企業(yè)和政府部門處理大規(guī)模數(shù)據(jù)必不可少的工具[7]。作為當(dāng)前最流行的NIO框架之一，Netty被廣泛應(yīng)用到物聯(lián)網(wǎng)、大數(shù)據(jù)分布式組件、通訊行業(yè)等，很多開源社區(qū)的開源組件也基于Netty的NIO框架進(jìn)行構(gòu)建。Netty主要是基于主從Reactors多線程模型，原理圖如圖1所示。簡(jiǎn)單來說，BossGroup負(fù)責(zé)請(qǐng)求接受，并將其注冊(cè)到WorkerGroup，然后WorkerGroup中的select監(jiān)聽事件，有事件就發(fā)送消息給下面的管道，管道里面有多線程可以同時(shí)處理很多業(yè)務(wù)，處理完返回。

圖1 Netty原理圖Figure 1 Schematic diagram of Netty

1.2 Kafka消息隊(duì)列

Kafka是一個(gè)高吞吐、分布式、基于發(fā)布訂閱的消息系統(tǒng)。Kafka在消息保存時(shí)根據(jù)Topic進(jìn)行歸類，發(fā)送消息者稱為Producer，消息接收者稱為Consumer。此外Kafka集群由多個(gè)Kafka實(shí)例組成，每個(gè)實(shí)例（server）稱為broker。Kafka工作原理示意圖如圖2所示，生產(chǎn)者（Producer）發(fā)布消息到指定的主題（Topic）中，消費(fèi)者（Consumer）訂閱該消息主題，這樣生產(chǎn)者和消費(fèi)者就可以不受對(duì)方的約束，完成各自的工作任務(wù)，對(duì)相互依賴性較強(qiáng)的復(fù)雜的業(yè)務(wù)進(jìn)行解耦合，簡(jiǎn)化設(shè)計(jì)流程。

圖2 Kafka原理示意圖Figure 2 Schematic diagram of Kafka

2 系統(tǒng)設(shè)計(jì)

2.1 系統(tǒng)架構(gòu)設(shè)計(jì)

本文系統(tǒng)架構(gòu)設(shè)計(jì)如圖3所示。通訊設(shè)備一般指包含無線通訊模塊和GPS模塊的智能終端設(shè)備，該設(shè)備可以采集工程機(jī)械車輛總線上的工況信息，并通過無線通信（如4G5G）把數(shù)據(jù)信息上報(bào)到平臺(tái)層。“負(fù)載均衡”可以在IP層對(duì)海量接入設(shè)備進(jìn)行分發(fā)，均衡到多個(gè)“網(wǎng)關(guān)服務(wù)”中，以減少每個(gè)“網(wǎng)關(guān)服務(wù)”承載的壓力?！熬W(wǎng)關(guān)服務(wù)”采用Netty技術(shù)框架，充分利用了Netty編解碼效率高和異步事件驅(qū)動(dòng)的特性，對(duì)上行數(shù)據(jù)進(jìn)行數(shù)據(jù)解碼、數(shù)據(jù)粘包處理，并在鏈路處理上增加數(shù)據(jù)過期機(jī)制，即對(duì)同一條鏈路添加等待超時(shí)處理邏輯，超過規(guī)定的時(shí)間（如60s）沒有接收到客戶端發(fā)來的數(shù)據(jù)包，則關(guān)閉超時(shí)鏈接以減小系統(tǒng)的負(fù)載[8-10]；“設(shè)備認(rèn)證”模塊根據(jù)協(xié)議和業(yè)務(wù)要求對(duì)不合法的接入設(shè)備進(jìn)行過濾，對(duì)合法的數(shù)據(jù)發(fā)送到下一步的業(yè)務(wù)模塊進(jìn)行處理。

圖3 系統(tǒng)總體架構(gòu)Figure 3 Overall system architecture

2.2 車聯(lián)網(wǎng)消息緩存設(shè)計(jì)

工程機(jī)械車聯(lián)網(wǎng)數(shù)據(jù)一般包含車輛的定位信息（如經(jīng)緯度、海拔）、車輛的工況數(shù)據(jù)（如油耗、吊重、臂長(zhǎng)等），當(dāng)車輛正式施工或者發(fā)生故障時(shí)，車載終端會(huì)以秒級(jí)甚至毫秒級(jí)高頻次上傳工況信息或故障信息。為了提高數(shù)據(jù)接入的性能而又不影響業(yè)務(wù)，如圖4所示，本文在數(shù)據(jù)處理上采用Kafka發(fā)布訂閱機(jī)制并結(jié)合Redis高性能緩存機(jī)制設(shè)計(jì)，目的在于盡量減少Netty的IO操作，最大化減少系統(tǒng)資源的開銷[10]。當(dāng)業(yè)務(wù)操作引發(fā)數(shù)據(jù)信息更改時(shí)，一方面系統(tǒng)啟動(dòng)定時(shí)線程去輪詢業(yè)務(wù)庫中的信息變更，更新到Redis緩存中，另一方面增加Kafka Producer監(jiān)控業(yè)務(wù)操作，一旦消息變更，Kafka Consumer會(huì)立即訂閱該變更信息，并更新到Redis緩存中。其中Redis緩存信息的數(shù)據(jù)結(jié)構(gòu)采用設(shè)備ID為主鍵的Map對(duì)象，保證緩存中設(shè)備數(shù)據(jù)的唯一性。這樣既能保證Redis緩存中業(yè)務(wù)信息的完整性，又能保證業(yè)務(wù)操作的信息變更能立即同步到緩存中。緩存中的信息將與設(shè)備上傳的信息共同組合參與后面的業(yè)務(wù)處理。

圖4 消息緩存設(shè)計(jì)原理圖Figure 4 Message cache design schematic

2.3 基于Kafka的業(yè)務(wù)解耦設(shè)計(jì)

工程機(jī)械車聯(lián)網(wǎng)數(shù)據(jù)信息一般包含兩種：一種為車聯(lián)網(wǎng)的業(yè)務(wù)信息，如車輛識(shí)別號(hào)、智能終端編號(hào)、司機(jī)信息、車輛所屬機(jī)構(gòu)等，業(yè)務(wù)處理方面，系統(tǒng)將根據(jù)實(shí)際需求進(jìn)行邏輯計(jì)算，如工況報(bào)警、圍欄報(bào)警等；另外一種為工程車輛工況信息，該信息一般為動(dòng)態(tài)信息，即從工程車輛采集的信息，如發(fā)動(dòng)機(jī)轉(zhuǎn)速、油耗、工作時(shí)間等，會(huì)隨時(shí)間的變化而變化。由于Kafka的Consumer線程直接執(zhí)行的業(yè)務(wù)操作耗時(shí)較多，會(huì)阻塞Consumer線程，若Consumer在一定時(shí)間范圍內(nèi)消費(fèi)不完這些數(shù)據(jù)，會(huì)判為提交失敗，Kafka的可靠機(jī)制會(huì)對(duì)這些數(shù)據(jù)進(jìn)行重復(fù)消費(fèi)，形成惡性循環(huán)，數(shù)據(jù)積累越來越嚴(yán)重，大大影響Netty整體處理數(shù)據(jù)的效率[11]。本方案的設(shè)計(jì)理念在于將業(yè)務(wù)處理模塊和車輛工況信息采集模塊進(jìn)行解耦，以防止局部處理過程中的阻塞影響整個(gè)數(shù)據(jù)處理流程。本文利用Kafka發(fā)布訂閱機(jī)制，在Netty架構(gòu)里只實(shí)現(xiàn)Kafka發(fā)布功能，訂閱功能在業(yè)務(wù)處理中執(zhí)行，實(shí)現(xiàn)數(shù)據(jù)接入和業(yè)務(wù)數(shù)據(jù)處理的線程解耦，耗時(shí)的復(fù)雜計(jì)算可以放在成熟的流式計(jì)算中處理，如Storm、Flink框架[12]。如圖5所示，Worker線程中使用Kafka的Producer把網(wǎng)關(guān)接入的數(shù)據(jù)對(duì)象緩存到Kafka消息隊(duì)列中，Kafka的Consumer Group根據(jù)Topic中獲取的數(shù)據(jù)對(duì)象，訂閱該隊(duì)列中的消息，并進(jìn)行實(shí)際業(yè)務(wù)邏輯處理，如報(bào)警服務(wù)、圖形展示、實(shí)時(shí)信息列表展示等。

圖5 業(yè)務(wù)解耦原理圖Figure 5 Service decoupling schematic

3 實(shí)驗(yàn)分析

3.1 實(shí)驗(yàn)環(huán)境

本文實(shí)驗(yàn)環(huán)境如表1所示，其中網(wǎng)關(guān)服務(wù)器部署了Zookeeper，版本為3.4.6.2.5，Kafka版本為0.10.0.2.5，MySQL版本為5.7.16，JDK版本為1.7，測(cè)試服務(wù)均使用java代碼編寫，兩個(gè)客戶端服務(wù)器模擬發(fā)送報(bào)文，三個(gè)服務(wù)器在同一局域網(wǎng)中，網(wǎng)絡(luò)配置千兆帶寬。

表1 實(shí)驗(yàn)環(huán)境Table 1 Experimental environment

3.2 實(shí)驗(yàn)設(shè)計(jì)及結(jié)果分析

為驗(yàn)證本文方案的有效性，選取傳統(tǒng)的Netty和MySQL方案記為方案1，Netty和Kafka（包含producer和consumer）記為方案2，與本文基于Netty結(jié)合Redis緩存并采取Kafka業(yè)務(wù)解耦的方案進(jìn)行對(duì)比測(cè)試。從服務(wù)端響應(yīng)耗時(shí)和吞吐量?jī)蓚€(gè)維度進(jìn)行測(cè)試和結(jié)果分析。實(shí)驗(yàn)中從實(shí)際車聯(lián)網(wǎng)設(shè)備選取上行報(bào)文，并采用模擬客戶端把該報(bào)文向服務(wù)端發(fā)送，并發(fā)數(shù)分別選取1000~10000，記錄服務(wù)端響應(yīng)耗時(shí)和吞吐量，實(shí)驗(yàn)多次取平均值。

從實(shí)驗(yàn)結(jié)果可以看出，如圖6所示，方案1由于沒有使用緩存和中間件，當(dāng)并發(fā)數(shù)超過3000時(shí)，由于大量的IO操作，系統(tǒng)資源消耗過快，開始出現(xiàn)數(shù)據(jù)堵塞現(xiàn)象，時(shí)間消耗開始大幅度增長(zhǎng)。方案2在并發(fā)數(shù)小于3000時(shí)，由于Consumer提交會(huì)有一定的時(shí)間消耗，所以優(yōu)勢(shì)不大，但當(dāng)并發(fā)數(shù)超過3000后，因復(fù)雜的耗時(shí)操作交由Kafka內(nèi)存中處理，縮短了響應(yīng)時(shí)間；當(dāng)并發(fā)數(shù)超過5000以后，方案2由于Consumer處理消息的速度已經(jīng)趕不上數(shù)據(jù)上傳的速度，也出現(xiàn)數(shù)據(jù)擁堵現(xiàn)象。本文方案在IO訪問上加入了Redis緩存和Kafka發(fā)布訂閱解耦的機(jī)制，彌補(bǔ)了前兩種方案的不足，數(shù)據(jù)處理耗時(shí)方面表現(xiàn)最優(yōu)。

圖6 系統(tǒng)耗時(shí)Figure 6 System time consumption

系統(tǒng)吞吐量方面，如圖7所示，當(dāng)并發(fā)量達(dá)到5000時(shí)，方案1吞吐量開始下降，表明此時(shí)大量的IO操作已經(jīng)影響到了系統(tǒng)吞吐量；當(dāng)并發(fā)量達(dá)到7000時(shí)，方案2吞吐量也出現(xiàn)了瓶頸，此時(shí)消費(fèi)開始出現(xiàn)堆積現(xiàn)象；本文方案在并發(fā)數(shù)達(dá)到10000時(shí)，系統(tǒng)依舊穩(wěn)定運(yùn)行。

圖7 系統(tǒng)吞吐量Figure 7 System Throughput

4 結(jié)語

本文提出了一種基于Netty的車聯(lián)網(wǎng)高性能數(shù)據(jù)采集系統(tǒng)設(shè)計(jì)方案。首先分析了傳統(tǒng)方案設(shè)計(jì)環(huán)節(jié)存在的弊端以及瓶頸產(chǎn)生的具體原因，然后從緩存和解耦兩個(gè)方面提出了系統(tǒng)的性能提升方案，最后通過實(shí)驗(yàn)驗(yàn)證了本方案在高并發(fā)情況下耗時(shí)和吞吐量?jī)蓚€(gè)方面的優(yōu)越性。本方案已經(jīng)在實(shí)際的工程機(jī)械車聯(lián)網(wǎng)行業(yè)投入生產(chǎn)使用，系統(tǒng)穩(wěn)定運(yùn)行且效果較好。