，，，，

(許繼電氣股份有限公司，河南 許昌 461000)

0 引言

消息總線服務(wù)為進(jìn)程間(多臺(tái)計(jì)算機(jī)之間和一臺(tái)計(jì)算機(jī)內(nèi)部)的消息傳輸提供支持，具有消息主題的創(chuàng)建 / 刪除、消息的訂閱 / 發(fā)布等功能，以接口函數(shù)的形式提供給各類應(yīng)用；具有組播、廣播和點(diǎn)到點(diǎn)傳輸形式，支持一對(duì)多、一對(duì)一的信息交換場(chǎng)合[1]。當(dāng)發(fā)送者需要分發(fā)一組新的消息時(shí)，它向消息總線服務(wù)器發(fā)出一個(gè)請(qǐng)求，消息總線服務(wù)器會(huì)負(fù)責(zé)把請(qǐng)求遞送給所有的訂閱者。經(jīng)過對(duì)當(dāng)前各種開源軟件和其它行業(yè)軟件的分析對(duì)比， 配電系統(tǒng)決定采用 Kafka 作為消息總線的主體服務(wù)。

消息總線服務(wù)為配電主站系統(tǒng)的各個(gè)模塊提供基于 topic 的消息訂閱和發(fā)布，消息總線服務(wù)應(yīng)考慮系統(tǒng)的集群部署、系統(tǒng)冗余功能等[2]；消息總線服務(wù)應(yīng)具備足夠的吞吐量，處理大量消息傳遞能力， 所以較高的性能是消息總線的基本要求。

1 消息總線實(shí)現(xiàn)機(jī)制

消息總線采用 Qt/C++ 對(duì) Kafka 的客戶端進(jìn)行封裝，屏蔽各種網(wǎng)絡(luò)處理細(xì)節(jié)，給平臺(tái)其它應(yīng)用一個(gè)簡(jiǎn)單易用的接口[3]。整個(gè)消息處理組件主要包括兩部分內(nèi)容：消息的發(fā)布和消息的處理。為了簡(jiǎn)化應(yīng)用，消息體現(xiàn)為函數(shù)參數(shù)，可以任意定義消息體結(jié) 構(gòu)而不改變程序結(jié)構(gòu)。消息的處理采用回調(diào)函數(shù)的方式，向系統(tǒng)注冊(cè)處理某種 topic 的回調(diào)函數(shù)即可。 下面具體介紹一下消息發(fā)布和消息處理的實(shí)現(xiàn)機(jī)制。

1.1 消息發(fā)布機(jī)制

消息發(fā)布機(jī)制是對(duì)kafka客戶端提供的信息發(fā)送函數(shù)進(jìn)行封裝，以達(dá)到向服務(wù)器上傳數(shù)據(jù)的目的[4]。由于事先不確定用戶上送的數(shù)據(jù)類型，封裝采用模板模式，支持任意數(shù)據(jù)類型。上送方式為單條上送，即一次只能上送一條數(shù)據(jù)，數(shù)據(jù)大小不做限制。并且接收服務(wù)器返回的上送數(shù)據(jù)狀態(tài)，成功返回0，失敗返回-1等。

1.2 消息處理機(jī)制

消息處理采用的是向服務(wù)器注冊(cè)機(jī)制，對(duì)kafka客戶端提供的注冊(cè)函數(shù)進(jìn)行封裝，以達(dá)到向服務(wù)器訂閱數(shù)據(jù)的目的。封裝技術(shù)與發(fā)布機(jī)制類似，采用模板形式。注冊(cè)可以分組注冊(cè)也可以同一組批量注冊(cè)。同一用戶可以注冊(cè)到服務(wù)器的不同組，但不能在同一組內(nèi)多次注冊(cè)。

2 消息總線性能指標(biāo)及測(cè)試方法分析

消息總線的性能指標(biāo)主要包括：大數(shù)據(jù)容量、數(shù)據(jù)處理速率、消費(fèi)者個(gè)數(shù)上限、消費(fèi)者個(gè)數(shù)上限、穩(wěn)定性等性能指標(biāo)。以上幾項(xiàng)性能指標(biāo)中大數(shù)據(jù)容量、數(shù)據(jù)處理速率、消費(fèi)者個(gè)數(shù)上限和穩(wěn)定性是最重要的，接下來我們具體分析這兩項(xiàng)性能指標(biāo)。

2.1 大數(shù)據(jù)容量

大數(shù)據(jù)容量分為單機(jī)數(shù)據(jù)容量和集群數(shù)據(jù)容量，是指消息總線的數(shù)據(jù)吞吐量[5]。主要測(cè)試方法是采用控制變量的方法將測(cè)試用例中的消息數(shù)量和單個(gè)消息大小分別增大，將數(shù)據(jù)總量逐級(jí)增加，測(cè)試容量上限，記錄上限值與單機(jī)最大容量是否成正比例。在上限范圍內(nèi)，單機(jī)或集群各項(xiàng)功能正常工作。

2.2 數(shù)據(jù)處理速率

數(shù)據(jù)處理速率又包括：生產(chǎn)者數(shù)據(jù)處理速率、消費(fèi)者數(shù)據(jù)處理速率、網(wǎng)絡(luò)風(fēng)暴中單條生產(chǎn)者生產(chǎn)速率和網(wǎng)絡(luò)風(fēng)暴中單條消費(fèi)者接收速率。

生產(chǎn)者數(shù)據(jù)處理速率是指生產(chǎn)者生產(chǎn)數(shù)據(jù)的效率。主要測(cè)試方法是啟用測(cè)試用例，以50%容量上限值發(fā)送數(shù)據(jù)，記錄Producer的生產(chǎn)時(shí)間，計(jì)算生產(chǎn)者的數(shù)據(jù)處理速率；更改測(cè)試用例，逐步提升到80%容量上限值發(fā)送量，記錄Producer生產(chǎn)時(shí)間，計(jì)算生產(chǎn)者數(shù)據(jù)處理速率，直至發(fā)送量提升至容量上限值，借此測(cè)試生產(chǎn)者數(shù)據(jù)處理速率上限。在上限范圍內(nèi)，集群各項(xiàng)功能正常工作。

消費(fèi)者數(shù)據(jù)處理速率指消費(fèi)者的接收效率。主要測(cè)試方法是啟用測(cè)試用例，以50%容量上限值發(fā)送數(shù)據(jù)，記錄Consumer的消費(fèi)時(shí)間，計(jì)算消費(fèi)者的數(shù)據(jù)處理速率；更改測(cè)試用例，逐步提升到80%容量上限值發(fā)送量，記錄Consumer消費(fèi)時(shí)間，計(jì)算消費(fèi)者數(shù)據(jù)處理速率，直至發(fā)送量提升至容量上限值，借此測(cè)試消費(fèi)者數(shù)據(jù)處理速率上限。在上限范圍內(nèi)，集群各項(xiàng)功能正常工作。

網(wǎng)絡(luò)風(fēng)暴中單條生產(chǎn)者生產(chǎn)速率指網(wǎng)絡(luò)風(fēng)暴情況下生產(chǎn)者生產(chǎn)一條數(shù)據(jù)的效率[6]。主要測(cè)試方法是更改測(cè)試用例，逐步提升到80%容量上限值發(fā)送量，記錄Producer生產(chǎn)時(shí)間，計(jì)算生產(chǎn)者單條數(shù)據(jù)處理速率，直至發(fā)送量提升至容量上限值，借此測(cè)試生產(chǎn)者單條數(shù)據(jù)處理速率上限。在上限范圍內(nèi)，集群各項(xiàng)功能正常工作。

網(wǎng)絡(luò)風(fēng)暴中單條消費(fèi)者接收速率指網(wǎng)絡(luò)風(fēng)暴情況下消費(fèi)者接收一條數(shù)據(jù)的效率。主要測(cè)試方法是更改測(cè)試用例，逐步提升到80%容量上限值發(fā)送量，記錄Consumer消費(fèi)時(shí)間，計(jì)算消費(fèi)者單條數(shù)據(jù)處理速率，直至發(fā)送量提升至容量上限值，借此測(cè)試消費(fèi)者單條數(shù)據(jù)處理速率上限。在上限范圍內(nèi)，集群各項(xiàng)功能正常工作。

2.3 消費(fèi)者個(gè)數(shù)上限

消費(fèi)者個(gè)數(shù)上限又分為單機(jī)消費(fèi)者個(gè)數(shù)上限和集群消費(fèi)者個(gè)數(shù)上限。其中單機(jī)消費(fèi)者上限是指單機(jī)消費(fèi)者的最大上限，主要測(cè)試方法是啟動(dòng)測(cè)試用例，使Producer以50%容量上限值持續(xù)生產(chǎn)數(shù)據(jù)，在一臺(tái)服務(wù)器上逐步增加Consumer個(gè)數(shù)同時(shí)消費(fèi)，測(cè)試消費(fèi)者個(gè)數(shù)上限；將測(cè)試用例調(diào)整至容量上限值，重復(fù)上述工作，測(cè)試消費(fèi)者上限個(gè)數(shù)。在上限范圍內(nèi)，各項(xiàng)功能正常工作。集群消費(fèi)者個(gè)數(shù)上限是指整個(gè)集群的消費(fèi)者上限者上限，主要測(cè)試方法是啟動(dòng)測(cè)試用例，使Producer以50%容量上限值持續(xù)生產(chǎn)數(shù)據(jù)，在多臺(tái)服務(wù)器上逐步增加Consumer個(gè)數(shù)同時(shí)消費(fèi)，測(cè)試消費(fèi)者總個(gè)數(shù)上限；將測(cè)試用例調(diào)整至容量上限值，重復(fù)上述工作，測(cè)試消費(fèi)者上限總個(gè)數(shù)，在上限范圍內(nèi)，集群各項(xiàng)功能正常工作。

2.4 穩(wěn)定性

穩(wěn)定性主要是指消息總線服務(wù)長(zhǎng)時(shí)間穩(wěn)定的運(yùn)行和在高負(fù)荷下穩(wěn)定的運(yùn)行。具體長(zhǎng)時(shí)間穩(wěn)定運(yùn)行是指最低保證50%容量上限值72小時(shí)穩(wěn)定運(yùn)行，具體測(cè)試方法是更改測(cè)試用例，50%容量上限值發(fā)送數(shù)據(jù)量，進(jìn)行72小時(shí)拷機(jī)測(cè)試，拷機(jī)期間各項(xiàng)功能正常，程序不應(yīng)產(chǎn)生異常或死機(jī)等現(xiàn)象，拷機(jī)測(cè)試期間觀測(cè)記錄CPU平均負(fù)荷率，拷機(jī)測(cè)試期間，觀測(cè)內(nèi)存占有率，不發(fā)生內(nèi)存泄露，拷機(jī)測(cè)試期間，各節(jié)點(diǎn)網(wǎng)絡(luò)平均負(fù)荷率。并且最低保證80%容量上限值30分鐘穩(wěn)定運(yùn)行，具體測(cè)試方法是更改測(cè)試用例，80%容量上限值發(fā)送數(shù)據(jù)量，進(jìn)行30分鐘拷機(jī)測(cè)試，拷機(jī)期間各項(xiàng)功能正常，程序不應(yīng)產(chǎn)生異?；蛩罊C(jī)等現(xiàn)象，拷機(jī)測(cè)試期間觀測(cè)記錄CPU平均負(fù)荷率，拷機(jī)測(cè)試期間，觀測(cè)記錄內(nèi)存占有率，不發(fā)生內(nèi)存泄露，拷機(jī)測(cè)試期間，記錄各節(jié)點(diǎn)網(wǎng)絡(luò)平均負(fù)荷率。高負(fù)荷穩(wěn)定運(yùn)行是指最低保證100%容量上限值5分鐘穩(wěn)定運(yùn)行，具體測(cè)試方法是更改測(cè)試用例，100%容量上限值發(fā)送數(shù)據(jù)量，進(jìn)行5分鐘壓力測(cè)試，壓力測(cè)試期間各項(xiàng)功能正常，程序不應(yīng)產(chǎn)生異?；蛩罊C(jī)等現(xiàn)象，拷機(jī)測(cè)試期間觀測(cè)記錄CPU平均負(fù)荷率，拷機(jī)測(cè)試期間，觀測(cè)記錄內(nèi)存占有率，不發(fā)生內(nèi)存泄露，拷機(jī)測(cè)試期間，記錄各節(jié)點(diǎn)網(wǎng)絡(luò)平均負(fù)荷率。

3 測(cè)試框架設(shè)計(jì)及實(shí)現(xiàn)

3.1 硬件架構(gòu)設(shè)計(jì)

消息總線測(cè)試環(huán)境采用三臺(tái)服務(wù)器搭載Kafka服務(wù)、組成集群，系統(tǒng)采用單網(wǎng)千兆網(wǎng)絡(luò)進(jìn)行連接,滿足2k+1配置，至少k+1臺(tái)服務(wù)器正常使用時(shí)，保證消息總線能正常工作。見圖1。

3.2 軟件框架設(shè)計(jì)

如圖2所示，消息總線軟件測(cè)試框架主要包括大數(shù)據(jù)生成、數(shù)據(jù)發(fā)送、數(shù)據(jù)接收和時(shí)間統(tǒng)計(jì)4個(gè)功能模塊。

圖2 消息總線性能測(cè)試軟件框架功能圖

數(shù)據(jù)生成模塊的主要功能是生成測(cè)試數(shù)據(jù)，為了便于統(tǒng)計(jì)數(shù)據(jù)大小，這里數(shù)據(jù)類型統(tǒng)一使用字符型?？梢詥蝹€(gè)添加數(shù)據(jù)，也可以批量生成大量數(shù)據(jù)。數(shù)據(jù)發(fā)送模塊主要功能是與消息總線服務(wù)器建立鏈接并發(fā)送數(shù)據(jù)到服務(wù)器。數(shù)據(jù)接收模塊主要功能是消費(fèi)者注冊(cè)和接收數(shù)據(jù)。時(shí)間統(tǒng)計(jì)模塊主要功能是統(tǒng)計(jì)并記錄消息總線處理數(shù)據(jù)的時(shí)間。由于要處理海量的數(shù)據(jù)，這里采用多線程技術(shù)，來保證測(cè)試框架界面的操作流暢。

3.3 軟件框架實(shí)現(xiàn)

Linux系統(tǒng)上采用 Qt/C++實(shí)現(xiàn)軟件框架。主窗口類設(shè)計(jì)如下：

class PUITest: public QMainWindow

{

Q_OBJECT

public:

explicit PUITest(QWidget *parent = 0);

～PUITest();

int m_DowNum,m_DowNum_reg,outcnt,totalcnt,cb_cnt;

outputDockWidget *m_InfoOutput;//操作信息輸出窗口

sendThread *sendmsgthread; //數(shù)據(jù)發(fā)送線程

runthread *runmsgthread; //海量數(shù)據(jù)發(fā)送線程

bool runOverflag;

int tId;//線程ID

void removeDockTitle(QDockWidget *docwidget);//去除頭標(biāo)志

void timerEvent(QTimerEvent *event);//定時(shí)器自動(dòng)調(diào)用函數(shù)

void InitTabWidget();

int test2(const QString str);

void Output(QString s);

void msgsendThread();

void magrunThread();

public slots:

void sendmsgover();

private slots:

void on_action_about_triggered();

void on_action_add_triggered();

void on_action_delete_triggered();

void n_action_reg_triggered();

void on_action_send_triggered();

void on_checkBox_clicked();

void on_action_unreg_triggered();

void on_action_addNumbers_triggered();

void on_action_save_triggered();

void on_action_open_triggered();

void on_action_run_triggered();

private:

Ui::PUITest *ui;

};

線程類設(shè)計(jì)如下：

class sendThread: public QThread

{

Q_OBJECT

signals:

void sendover();//發(fā)送數(shù)據(jù)結(jié)束信號(hào)

protected:

void run();//線程自動(dòng)調(diào)用

};

框架實(shí)現(xiàn)主界面如圖3所示，主界面中有兩個(gè)分頁面，一個(gè)是發(fā)送信息配置，主要包括數(shù)據(jù)生成、發(fā)送次數(shù)設(shè)置、發(fā)送延時(shí)設(shè)置等。一個(gè)是接收信息配置，主要是注冊(cè)信息的設(shè)置等。數(shù)據(jù)可以單個(gè)添加，更可以批量生成。另外還有操作提示窗口，時(shí)間統(tǒng)計(jì)窗口等。

圖3 消息總線測(cè)試軟件框架主界面

4 消息總線性能測(cè)試及效果分析

4.1 性能測(cè)試

采用前幾節(jié)中提出的測(cè)試方法及設(shè)計(jì)的測(cè)試框架，對(duì)消息總線性能進(jìn)行測(cè)試。測(cè)試中使用三臺(tái)服務(wù)器組成消息總線集群，兩臺(tái)客戶端安裝測(cè)試程序。具體軟硬件信息如表1和表2所示。

表1 軟硬件配置環(huán)境

表2 軟件消息

4.2 測(cè)試效果

通過測(cè)試框架的測(cè)試軟件自動(dòng)生成海量的測(cè)試數(shù)據(jù)，并自動(dòng)統(tǒng)計(jì)數(shù)據(jù)處理時(shí)間和讀取服務(wù)器資源占用信息。以下所有記錄均在HP DL560 G8 服務(wù)器上進(jìn)行測(cè)試，測(cè)試10次求平均值，性能測(cè)試結(jié)果如表3～6所示。

表3 生產(chǎn)者生產(chǎn)速率

表4 消費(fèi)者消費(fèi)速率

表5 網(wǎng)絡(luò)風(fēng)暴中單條生產(chǎn)信息生產(chǎn)時(shí)間

表6 網(wǎng)絡(luò)風(fēng)暴中單條消費(fèi)消息收時(shí)間

表3是生產(chǎn)者生產(chǎn)效率的的測(cè)試結(jié)果，從該測(cè)試結(jié)果中可以發(fā)現(xiàn)，發(fā)送量一定的情況下，隨著單條數(shù)據(jù)字節(jié)數(shù)的增加，發(fā)送效率逐漸降低。表4是消費(fèi)者消費(fèi)效率的測(cè)試的測(cè)試結(jié)果，從該測(cè)試結(jié)果中可以發(fā)現(xiàn)，接收量一定的情況下，隨著單條數(shù)據(jù)字節(jié)數(shù)的增加，接收效率逐漸降低。通過對(duì)比發(fā)現(xiàn)，消費(fèi)效率要明顯高于生產(chǎn)效率。表5是網(wǎng)絡(luò)風(fēng)暴中單條信息的生產(chǎn)效率測(cè)試結(jié)果，從該測(cè)試結(jié)果中可以發(fā)現(xiàn)，不同程度網(wǎng)絡(luò)風(fēng)暴情況下，生產(chǎn)單條數(shù)據(jù)的效率相對(duì)穩(wěn)定。也就是說網(wǎng)絡(luò)風(fēng)暴對(duì)生產(chǎn)者的生產(chǎn)效率影響不大。表6是網(wǎng)絡(luò)風(fēng)暴中單條消費(fèi)信息的消費(fèi)效率的測(cè)試結(jié)果，從該測(cè)試結(jié)果中可以發(fā)現(xiàn)，不同程度網(wǎng)絡(luò)風(fēng)暴情況下，消費(fèi)單條數(shù)據(jù)的效率不穩(wěn)定。也就是說網(wǎng)絡(luò)風(fēng)暴對(duì)消費(fèi)者的效率影響較大。

消費(fèi)者上限測(cè)試，當(dāng)消費(fèi)者為10 000時(shí)，系統(tǒng)各項(xiàng)指標(biāo)均正常運(yùn)行。這以遠(yuǎn)遠(yuǎn)超出我們實(shí)際應(yīng)用時(shí)的上限，新型配電主站系統(tǒng)實(shí)際應(yīng)用時(shí)的消費(fèi)者數(shù)量不超過1 000，所以這里不再具體測(cè)試。

穩(wěn)定性測(cè)試時(shí)接入不少于 10 000臺(tái)配電終端，每臺(tái)終端定義100個(gè)“ 三遙” 數(shù)據(jù)信息接入被測(cè)系統(tǒng)，系統(tǒng)應(yīng)不間斷運(yùn)行維持 72 h， 被測(cè)系統(tǒng) cpu及內(nèi)存使用情況如下：

1)當(dāng)50%發(fā)送速率拷機(jī)，單條消息字節(jié)長(zhǎng)度為1 024 byte時(shí)，內(nèi)存占用率和網(wǎng)絡(luò)負(fù)荷負(fù)荷率正常；

2)當(dāng)80%發(fā)送速率拷機(jī)，單條消息字節(jié)長(zhǎng)度為1 024 byte時(shí)，內(nèi)存占用率和網(wǎng)絡(luò)負(fù)荷負(fù)荷率正常；

3)當(dāng)100%發(fā)送速率拷機(jī)，單條消息字節(jié)長(zhǎng)度為1 024 byte時(shí)，內(nèi)存占用率和網(wǎng)絡(luò)負(fù)荷負(fù)荷率較高。

經(jīng)性能測(cè)試，除了掌握以上性能規(guī)律外，還發(fā)現(xiàn)了兩個(gè)有價(jià)值的問題，問題1:當(dāng)向消息總線服務(wù)器發(fā)送海量數(shù)據(jù)，導(dǎo)致服務(wù)器存儲(chǔ)空間不足時(shí)，消息總線服務(wù)直接崩潰；問題2:向消息總線服務(wù)器長(zhǎng)時(shí)間發(fā)送1 024 byte大小的數(shù)據(jù)流時(shí)，服務(wù)器內(nèi)存占用過大。以上性能數(shù)據(jù)及問題得到了設(shè)計(jì)人員的高度認(rèn)可，充分體現(xiàn)了文中設(shè)計(jì)的測(cè)試框架的價(jià)值。

5 結(jié)論

隨著配電信息數(shù)據(jù)規(guī)模的不斷擴(kuò)大，對(duì)消息總線的性能要求越來越高，而數(shù)據(jù)吞吐量和數(shù)據(jù)處理效率是消息總線性能的重要指標(biāo)。本文通過對(duì)kafka為服務(wù)主體的消息總線性能進(jìn)行分析后，提出了效果較好的測(cè)試方法，設(shè)計(jì)了實(shí)用性很強(qiáng)的測(cè)試框架。經(jīng)實(shí)際應(yīng)用證明，該測(cè)試框架簡(jiǎn)單易用，在很大程度上提高了性能測(cè)試的效率。該框架有較好的通用性，其他消息總線的測(cè)試也可以使用。