基于Kafka的分布式能效管理平臺(tái)的設(shè)計(jì)與實(shí)現(xiàn)*

朱幼普 盧 軍

（武漢郵電科學(xué)研究院 武漢 430074）

1 引言

隨著互聯(lián)網(wǎng)技術(shù)的發(fā)展越來(lái)越迅速，通過(guò)信息化管理平臺(tái)對(duì)能源進(jìn)行分配和管理，實(shí)現(xiàn)能源的信息化監(jiān)管和控制，從而達(dá)到節(jié)約能源和保護(hù)環(huán)境的目的也越來(lái)越明確。如何將各種計(jì)量設(shè)備（水表，電表，傳感器等）采集的數(shù)據(jù)進(jìn)行實(shí)時(shí)可靠的傳輸，如何應(yīng)對(duì)計(jì)量設(shè)備數(shù)據(jù)準(zhǔn)實(shí)時(shí)訪問(wèn)的業(yè)務(wù)應(yīng)用需求、提高計(jì)量設(shè)備采集數(shù)據(jù)的準(zhǔn)實(shí)時(shí)應(yīng)用能力，以及對(duì)計(jì)量設(shè)備狀態(tài)的準(zhǔn)實(shí)時(shí)監(jiān)控，成為搭建能效信息化管理平臺(tái)需要考慮的關(guān)鍵問(wèn)題。Apache Kafka，作為一種分布式的消息系統(tǒng)，由于具有可水平擴(kuò)展、高吞吐率和實(shí)時(shí)性等優(yōu)點(diǎn)而被廣泛的使用。本文重點(diǎn)研究數(shù)據(jù)的傳輸過(guò)程，并搭建了一個(gè)基于Kafka分布式消息隊(duì)列的能效數(shù)據(jù)管理平臺(tái)，為機(jī)關(guān)或小區(qū)的分項(xiàng)用能數(shù)據(jù)的智能采集，狀態(tài)監(jiān)控、統(tǒng)計(jì)分析和能耗管理等提供了良好的支持。

2 相關(guān)技術(shù)介紹

2.1 Kafka分布式消息系統(tǒng)

Kafka是由Linked In研發(fā)的一個(gè)分布式消息系統(tǒng)，它以具有水平擴(kuò)展和高吞吐量率的特性而被廣泛使用。目前越來(lái)越多的開(kāi)源分布式處理系統(tǒng)，例如Storm、Spark都支持與Kafka集成［1］。

近年來(lái)，對(duì)流數(shù)據(jù)和運(yùn)營(yíng)數(shù)據(jù)的分析處理已經(jīng)成為在線分析、實(shí)時(shí)監(jiān)控等應(yīng)用的重要組成部分［2］。Kafka正是針對(duì)這樣的需求，將海量且復(fù)雜的數(shù)據(jù)進(jìn)行分布式緩存的消息系統(tǒng)［3］。其主要設(shè)計(jì)目標(biāo)如下：

1）高效的數(shù)據(jù)持久化能力。能夠在常數(shù)時(shí)間復(fù)雜度實(shí)現(xiàn)TB級(jí)以上的數(shù)據(jù)讀∕寫(xiě)入硬盤(pán)。

2）高吞吐率。在廉價(jià)的商用機(jī)器上單機(jī)可支持每秒100萬(wàn)條消息的讀寫(xiě)。

3）支持在Broker中進(jìn)行消息的分區(qū)存儲(chǔ)，并且對(duì)分區(qū)中消息的讀取是有序的。

4）在線也能實(shí)現(xiàn)水平擴(kuò)展。

Kafka分布式消息系統(tǒng)在收集和分發(fā)海量的日志數(shù)據(jù)文件時(shí)具有較低的延時(shí)性［4］。它可同時(shí)支持?jǐn)?shù)據(jù)的在線處理和離線處理。Kafka的系統(tǒng)設(shè)計(jì)（例如，分布式架構(gòu)，分區(qū)存儲(chǔ)，順序硬盤(pán)讀寫(xiě)等）使其在吞吐量和可伸縮性方面具有很好的表現(xiàn)。Linked In公司在使用Kafka一段時(shí)間后，每天的數(shù)據(jù)處理量可以達(dá)到百G級(jí)別。

Kafka的架構(gòu)如圖1所示。

圖1 Kafka架構(gòu)圖

本文的能效管理系統(tǒng)之所以選用Kafka作為實(shí)現(xiàn)各個(gè)微服務(wù)之間數(shù)據(jù)緩存與訂閱模塊主要由于其具有以下優(yōu)勢(shì)：

解耦合：各個(gè)微服務(wù)通過(guò)實(shí)現(xiàn)Kafka這個(gè)統(tǒng)一的接口進(jìn)行數(shù)據(jù)交換，降低了系統(tǒng)模塊之間的耦合。

可擴(kuò)展性：若采集微服務(wù)向Kafka推送的消息量劇增，可通過(guò)水平擴(kuò)展Kafka的broker節(jié)點(diǎn)實(shí)現(xiàn)消息量的負(fù)載均衡。

強(qiáng)大的緩沖能力：面對(duì)微服務(wù)之間訪問(wèn)數(shù)據(jù)量的劇增，Kafka消息隊(duì)列可以有效地緩沖系統(tǒng)的流量壓力，避免系統(tǒng)在大數(shù)據(jù)的壓力下崩潰，保證了系統(tǒng)的穩(wěn)定性［5］。

健壯性好：作為分布式消息系統(tǒng)，Kafka在部分功能失效的情況下，不會(huì)影響整個(gè)系統(tǒng)的正常運(yùn)行。

2.2 Zookeeper［6］

Zookeeper是一個(gè)高性能，開(kāi)源分布式應(yīng)用協(xié)調(diào)服務(wù)。分布式應(yīng)用可以基于它實(shí)現(xiàn)更高級(jí)的服務(wù)，比如同步，配置管理，集群管理，名空間。Zookeeper使用文件目錄樹(shù)作為數(shù)據(jù)模型。

針對(duì)學(xué)生整體英語(yǔ)基礎(chǔ)差、聽(tīng)力能力普遍較低、聽(tīng)力訓(xùn)練投入時(shí)間不足和大學(xué)英語(yǔ)聽(tīng)力教學(xué)課時(shí)銳減、大學(xué)英語(yǔ)四、六級(jí)考試聽(tīng)力分值比例增加的嚴(yán)峻現(xiàn)實(shí)，獨(dú)立學(xué)院的大學(xué)英語(yǔ)教師應(yīng)積極按照《大學(xué)英語(yǔ)教學(xué)指南》的要求，充分利用現(xiàn)代信息技術(shù)，為營(yíng)造多渠道、多方面的教學(xué)環(huán)境和學(xué)習(xí)環(huán)境；獨(dú)立學(xué)院非英語(yǔ)專業(yè)學(xué)生更應(yīng)積極配合老師，在信息技術(shù)環(huán)境下，積極進(jìn)行聽(tīng)力練習(xí)，來(lái)提高自身的英語(yǔ)聽(tīng)力能力。

由于Zookeeper可配置管理分布式系統(tǒng)，因此Kafka集群通常使用Zookeeper進(jìn)行協(xié)調(diào)管理。Kafka代理的增加和減少都將觸發(fā)Zookeeper服務(wù)通知生產(chǎn)者和消費(fèi)者，生產(chǎn)者和消費(fèi)者就會(huì)與Kafka集群中的其它代理協(xié)調(diào)工作。

3 基于Kafka分布式能效管理平臺(tái)的設(shè)計(jì)

3.1 系統(tǒng)的整體架構(gòu)設(shè)計(jì)

系統(tǒng)整體劃分為五個(gè)層次，從上到下依次為接入層、控制層、中間件服務(wù)層、微服務(wù)業(yè)務(wù)支撐層和數(shù)據(jù)存儲(chǔ)層。如圖2所示：

接入層：接入層由Nginx服務(wù)器集群組成，用戶請(qǐng)求首先發(fā)送到接入層，Nginx服務(wù)器接收到用戶請(qǐng)求后，通過(guò)負(fù)載均衡將請(qǐng)求反向代理到控制層［7］。

圖2 系統(tǒng)整體架構(gòu)設(shè)計(jì)圖

控制層：控制層使用Spring Boot作為開(kāi)發(fā)平臺(tái)。Spring Boot提供了開(kāi)箱即用的SpringMVC框架，使用SpringMVC處理接入層轉(zhuǎn)發(fā)過(guò)來(lái)的請(qǐng)求大致可分為三種情況：1）從Redis緩存中讀取數(shù)據(jù)；2）從MongoDB中獲取上傳的文件或圖片［8］；3）調(diào)用微服務(wù)接口執(zhí)行業(yè)務(wù)。另外，需要配合Redis緩存服務(wù)器實(shí)現(xiàn)session共享。

中間件服務(wù)器層：使用Zookeeper作為服務(wù)注冊(cè)配置中心，所有向外提供接口的微服務(wù)都需要在Zookeeper上進(jìn)行注冊(cè)后方可被其他消費(fèi)者調(diào)用?？刂茖釉谔幚碛脩粽?qǐng)求時(shí)，將作為服務(wù)接口的消費(fèi)者，在Zookeeper上查詢和執(zhí)行相應(yīng)的微服務(wù)。在Zookeeper之上配合Dubbox對(duì)微服務(wù)進(jìn)行管理和監(jiān)控。將Redis作為緩存服務(wù)器，一方面為控制層提供session共享支持，另一方面為分布式系統(tǒng)提供緩存服務(wù)，加快數(shù)據(jù)查詢響應(yīng)速度和效率［9］。將Kafka作為消息通信服務(wù)器，支撐各個(gè)微服務(wù)之間的通信，實(shí)現(xiàn)消息收集和傳播的功能。

微服務(wù)業(yè)務(wù)支持層［10～12］：

2）日志微服務(wù)：記錄用戶操作日志和系統(tǒng)日志，提供日志錄入和查詢的接口；

3）權(quán)限微服務(wù)：維護(hù)用戶、用戶組、角色、菜單、按鈕數(shù)據(jù)，提供鑒權(quán)的接口；

4）設(shè)備微服務(wù)：維護(hù)計(jì)量設(shè)備、電網(wǎng)環(huán)境、能耗設(shè)備、節(jié)點(diǎn)信息；

5）采集微服務(wù)：采集計(jì)量設(shè)備上報(bào)的實(shí)時(shí)數(shù)據(jù)，并將數(shù)據(jù)記錄到日示數(shù)表，配合Redis緩存支撐報(bào)表查詢業(yè)務(wù)；

6）報(bào)警微服務(wù)：維護(hù)報(bào)警等級(jí)、報(bào)警級(jí)別、報(bào)警類型和實(shí)時(shí)報(bào)警數(shù)據(jù)，配合Redis緩存支撐報(bào)警報(bào)表業(yè)務(wù)；

7）外部數(shù)據(jù)微服務(wù)：提供第三方數(shù)據(jù)的導(dǎo)入、存儲(chǔ)和查詢的接口，為報(bào)表提供數(shù)據(jù)支持；

8）報(bào)表微服務(wù)：提供報(bào)表數(shù)據(jù)獲取，分析和查詢；

9）分析微服務(wù)：對(duì)采集微服務(wù)采集的數(shù)據(jù)進(jìn)行處理，并將處理結(jié)果存入數(shù)據(jù)庫(kù)中。

3.2 基于Kafka的數(shù)據(jù)傳輸流程實(shí)現(xiàn)

本系統(tǒng)中，Kafka是微服務(wù)之間通信的媒介［13～14］，至關(guān)重要。計(jì)量設(shè)備采集的數(shù)據(jù)是我們需要分析利用的數(shù)據(jù)，這些數(shù)據(jù)存儲(chǔ)在專門(mén)的數(shù)據(jù)庫(kù)文件中，當(dāng)其他微服務(wù)（例如分析微服務(wù)或報(bào)警微服務(wù)）需要對(duì)這些數(shù)據(jù)進(jìn)行分析處理時(shí)，采集微服務(wù)作為消息的推送者（producer），會(huì)先將數(shù)據(jù)推送到Kafka集群中，Kafka集群則將這些數(shù)據(jù)以日志文件的形式持久化到硬盤(pán)中，分析微服務(wù)和報(bào)警微服務(wù)作為消費(fèi)者（consumer），會(huì)從Kafka消息隊(duì)列中拉取數(shù)據(jù)進(jìn)行消費(fèi)，消費(fèi)者對(duì)每一條消息的消費(fèi)情況會(huì)同步到Zookeeper服務(wù)器集群中，若某個(gè)微服務(wù)出現(xiàn)異常停止消費(fèi)，Zookeeper會(huì)將此消費(fèi)者最后消費(fèi)消息的偏移量offset記錄下來(lái)，等到微服務(wù)恢復(fù)正常后從此偏移量繼續(xù)消費(fèi)消息，從而確保每條消息被成功消費(fèi)。因此，本系統(tǒng)在數(shù)據(jù)傳輸方面的設(shè)計(jì)具有較高的可靠性。此外，本系統(tǒng)采用Kafka提供的數(shù)據(jù)批處理的接口，計(jì)量設(shè)備采集的數(shù)據(jù)批量傳輸?shù)終afka集群中，其他微服務(wù)從Kafka集群里批量拉取數(shù)據(jù)進(jìn)行分析處理，減少了網(wǎng)絡(luò)傳輸?shù)拈_(kāi)銷(xiāo)，使整個(gè)系統(tǒng)的性能更高。

4 實(shí)驗(yàn)驗(yàn)證與結(jié)果

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

實(shí)驗(yàn)環(huán)境所用服務(wù)器為高性能服務(wù)器，機(jī)器配置如表1所示。

測(cè)試主機(jī)有3臺(tái)，用于搭建Kafka和Zookeeper的集群，3臺(tái)主機(jī)網(wǎng)絡(luò)配置信息如表2所示。

表2 Kafka測(cè)試集群信息

4.2 實(shí)驗(yàn)過(guò)程及結(jié)果

啟動(dòng)Kafka集群和Zookeeper集群，準(zhǔn)備一定規(guī)模的源數(shù)據(jù)進(jìn)行采集，設(shè)定采集微服務(wù)向Kafka集群中推送消息的時(shí)間為每5分鐘一次，然后依次啟動(dòng)分析微服務(wù)，報(bào)警微服務(wù)等來(lái)消費(fèi)這些數(shù)據(jù)，通過(guò)Kafka自帶的監(jiān)控工具來(lái)查看數(shù)據(jù)消費(fèi)的情況。

實(shí)驗(yàn)結(jié)果如圖3所示，圖中描繪了對(duì)于不同數(shù)量的平行消費(fèi)者，消息的吞吐量變化的趨勢(shì)。圖中的虛線代表消費(fèi)者每秒鐘消費(fèi)的消息條數(shù)，實(shí)線代表消費(fèi)者每秒鐘消費(fèi)的消息數(shù)據(jù)量的大小，從圖中可以清晰地看出，隨著平行消費(fèi)者數(shù)量的增加，數(shù)據(jù)的吞吐量基本呈線性增加。

圖3 消息吞吐量

5 結(jié)語(yǔ)

作為新一代分布式消息系統(tǒng)，在大數(shù)據(jù)背景下的今天，Kafka為我們處理海量數(shù)據(jù)提供了研究方向。本文利用Kafka在數(shù)據(jù)傳輸方面的優(yōu)勢(shì)，設(shè)計(jì)了一個(gè)實(shí)時(shí)可靠的能效信息管理平臺(tái)，確保了設(shè)備采集數(shù)據(jù)傳輸?shù)膶?shí)時(shí)性和可靠性，而且采用Kafka分布式部署的方式，提高了平臺(tái)的可擴(kuò)展性。然而，在某些極端情況下，本系統(tǒng)也存在缺陷，當(dāng)采集微服務(wù)持續(xù)不斷地向Kafka集群中推送消息時(shí)，會(huì)導(dǎo)致Kafka需要保存的數(shù)據(jù)量非常大，整個(gè)系統(tǒng)的性能會(huì)受到影響［15］。針對(duì)此問(wèn)題，在下一步研究中，我們還需要對(duì)系統(tǒng)進(jìn)行更深入的優(yōu)化，使系統(tǒng)具有更好的性能。