馮 娟

（上海申通地鐵集團(tuán)有限公司技術(shù)中心，201103，上?！胃呒?jí)工程師）

ZooKeeper在復(fù)雜事件處理系統(tǒng)中的應(yīng)用

馮 娟

（上海申通地鐵集團(tuán)有限公司技術(shù)中心，201103，上?！胃呒?jí)工程師）

復(fù)雜事件的分布式計(jì)算處理與協(xié)調(diào)一直是分布式計(jì)算的難題，尤其是基于低成本的高可用性技術(shù)實(shí)踐。提出了把Zoo Keeper應(yīng)用于以Esper為核心的復(fù)雜事件處理系統(tǒng)的一種技術(shù)解決方案。該方案通過Zoo Keeper協(xié)調(diào)多個(gè)復(fù)雜事件處理引擎，并作為控制臺(tái)和復(fù)雜事件處理引擎的交互媒介，能夠?qū)崿F(xiàn)復(fù)雜事件處理分布式計(jì)算的集群技術(shù)應(yīng)用，可提高系統(tǒng)的可用性和實(shí)時(shí)性。所提出的方案在3臺(tái)PC機(jī)搭建的仿真實(shí)驗(yàn)平臺(tái)進(jìn)行了驗(yàn)證，說明是可行的。

自動(dòng)售檢票系統(tǒng)；設(shè)備監(jiān)控；數(shù)據(jù)流處理；分布式計(jì)算；集群技術(shù)；

Author'saddressTechnology Center of Shanghai Shentong Metro Group Co.，Ltd.，201103，Shanghai，China

敏捷企業(yè)的一個(gè)基本要求就是事件驅(qū)動(dòng)業(yè)務(wù)（Event-driven Business）。事件驅(qū)動(dòng)業(yè)務(wù)就是在連續(xù)業(yè)務(wù)過程中進(jìn)行決策的一種業(yè)務(wù)管理方式，即根據(jù)不同時(shí)間出現(xiàn)的一系列事件而執(zhí)行相關(guān)任務(wù)，并調(diào)度可用的資源執(zhí)行任務(wù)。在軟件開發(fā)中，事件驅(qū)動(dòng)編程能夠帶來強(qiáng)大的功能和更靈活的交互性，而在企業(yè)中，事件驅(qū)動(dòng)業(yè)務(wù)依托于比較成熟的信息化建設(shè)，能夠?qū)I(yè)務(wù)的靈活性和效率大幅度提高。在各個(gè)業(yè)務(wù)應(yīng)用系統(tǒng)連續(xù)不斷地產(chǎn)生數(shù)據(jù)流的同時(shí)，根據(jù)預(yù)先定義的條件生成一些“業(yè)務(wù)事件”，按照策略對(duì)這些業(yè)務(wù)事件進(jìn)行分析處理，觸發(fā)新的業(yè)務(wù)事件或者業(yè)務(wù)流程，即實(shí)現(xiàn)了業(yè)務(wù)的事件驅(qū)動(dòng)［1］。事件驅(qū)動(dòng)業(yè)務(wù)要求能夠快速（毫秒級(jí)）、不間斷地處理連續(xù)、海量的數(shù)據(jù)，具備靈活的規(guī)則或策略設(shè)置，從而具備迅速識(shí)別、捕獲、響應(yīng)實(shí)時(shí)業(yè)務(wù)數(shù)據(jù)的能力。而傳統(tǒng)的企業(yè)IT架構(gòu)通常在業(yè)務(wù)應(yīng)用系統(tǒng)中處理業(yè)務(wù)操作遵循固定的業(yè)務(wù)流程處理跨系統(tǒng)事務(wù)，并且這些流程很少變化，基于數(shù)據(jù)倉(cāng)庫(kù)進(jìn)行海量數(shù)據(jù)的存儲(chǔ)及事后分析的這種IT架構(gòu)遠(yuǎn)遠(yuǎn)達(dá)不到事件驅(qū)動(dòng)業(yè)務(wù)的要求［2］。事件驅(qū)動(dòng)業(yè)務(wù)能夠應(yīng)用的業(yè)務(wù)領(lǐng)域很多，凡是需要快速處理連續(xù)性數(shù)據(jù)、需要能夠靈活制定策略的業(yè)務(wù)，都可以采用事件驅(qū)動(dòng)的業(yè)務(wù)模式。例如，軌道交通票務(wù)數(shù)據(jù)在線交易處理、軌道交通AFC（自動(dòng)售檢票）系統(tǒng)設(shè)備運(yùn)行狀態(tài)實(shí)時(shí)監(jiān)控、證券行業(yè)常見的風(fēng)險(xiǎn)分析及預(yù)警（事前及事中風(fēng)險(xiǎn)控制）、投資決策中的程序化交易、經(jīng)紀(jì)人績(jī)效計(jì)算等。

1 ZooKeeper技術(shù)

Zoo Keeper是Hadoop的正式子項(xiàng)目，是一種用于提供配置信息維護(hù)、命名服務(wù)、分布式同步和組服務(wù)等的集中式協(xié)調(diào)系統(tǒng)，所有這些服務(wù)都以一種或多種形式被分布式系統(tǒng)使用［3-4］。

Zoo Keeper本質(zhì)上是一個(gè)分布式的小文件存儲(chǔ)系統(tǒng)，具有下述應(yīng)用特性：

（1）簡(jiǎn)單。Zoo Keeper允許通過分布式程序共享方式，組織成類似標(biāo)準(zhǔn)文件系統(tǒng)層級(jí)命名空間協(xié)調(diào)的分布式程序。這個(gè)命名空間包含類似文件和目錄的數(shù)據(jù)寄存器，不像一個(gè)典型的用于存儲(chǔ)的文件系統(tǒng)。Zoo Keeper的數(shù)據(jù)保存在內(nèi)存中，可實(shí)現(xiàn)高吞吐量和低延遲訪問。

（2）冗余。ZooKeeper本身和其自身協(xié)調(diào)的程序一樣，設(shè)計(jì)為冗余的，在擁有許多主機(jī)集合的主機(jī)組上運(yùn)行。組成Zoo Keeper服務(wù)群的所有服務(wù)器都已知對(duì)方，只要所有服務(wù)器中的主服務(wù)器可用，ZooKeeper服務(wù)就是可用的?？蛻舳随溄拥囊粋€(gè)Zoo Keeper服務(wù)器通過發(fā)送請(qǐng)求、獲取響應(yīng)、監(jiān)聽事件、發(fā)送心跳維持一個(gè)TCP（傳輸控制協(xié)議）鏈接，如果鏈接中斷，客戶端將連接到另外一個(gè)服務(wù)器［5］。

（3）有序。Zoo Keeper用一個(gè)反映所有Zoo Keeper事務(wù)順序的數(shù)字，標(biāo)記每一個(gè)更新。以后的操作可以使用這個(gè)順序數(shù)字標(biāo)記。

（4）快速。ZooKeeper在讀取占主要地位的負(fù)載環(huán)境下，在讀取操作比寫入操作更頻繁的情況下，讀取速度非?？?。

（5）豐富的API（應(yīng)用程序編程接口）。ZooKeeper為開發(fā)人員提供了一套豐富的API，減輕了開發(fā)人員編寫通用協(xié)議的負(fù)擔(dān)［6］。

2 構(gòu)建ZooKeeper的軟件系統(tǒng)架構(gòu)

在傳統(tǒng)的IT架構(gòu)中，已經(jīng)實(shí)現(xiàn)了業(yè)務(wù)事件的處理。比如在傳統(tǒng)的軌道交通自動(dòng)售檢票系統(tǒng)的票務(wù)處理中，通常將票務(wù)交易數(shù)據(jù)經(jīng)檢驗(yàn)和合法性判斷后存儲(chǔ)在聯(lián)機(jī)數(shù)據(jù)庫(kù)中，進(jìn)一步的數(shù)據(jù)分析需在軌道交通晚間停運(yùn)后或通過應(yīng)用系統(tǒng)的操作界面由既定的報(bào)表處理模塊完成各類業(yè)務(wù)報(bào)表或響應(yīng)聯(lián)機(jī)查詢。又如在AFC系統(tǒng)中，所有的AFC終端設(shè)備自動(dòng)根據(jù)預(yù)設(shè)定時(shí)間間隔，把大量設(shè)備自檢狀態(tài)數(shù)據(jù)發(fā)送至線路中央計(jì)算機(jī)系統(tǒng)或票務(wù)清分中心。為實(shí)時(shí)監(jiān)視和控制AFC終端設(shè)備，需要處理程序快速鑒別設(shè)備狀態(tài)的海量數(shù)據(jù)及其與上一幀設(shè)備運(yùn)行監(jiān)視屏數(shù)據(jù)的差異，若存有差異需要實(shí)時(shí)自動(dòng)刷新設(shè)備運(yùn)行監(jiān)視屏信息［7］。上述兩個(gè)應(yīng)用舉例，都存在著CEP（復(fù)雜事件處理）需求。傳統(tǒng)的CEP方式通常存在2個(gè)問題：一是速度慢，大量復(fù)雜事件的處理順序是基于交易數(shù)據(jù)寫入數(shù)據(jù)庫(kù)后的后續(xù)處理順序確定，易造成數(shù)據(jù)訪問的I/O瓶頸，處理效率低；二是對(duì)于復(fù)雜或異常情況的處理，需要人工介入處理，操作人員的經(jīng)驗(yàn)很大程度上確定了復(fù)雜事件的處理方法和效果。如AFC終端設(shè)備上傳的票務(wù)交易數(shù)據(jù)發(fā)生飄移和發(fā)現(xiàn)某AFC終端設(shè)備運(yùn)行狀態(tài)異常，相應(yīng)的后續(xù)處理步驟都需要人工介入。解決這兩個(gè)問題的有效技術(shù)方法，需要松耦合的多處理機(jī)（服務(wù)器）集群技術(shù)解決方案，以部署復(fù)雜事件并行處理的協(xié)調(diào)步驟并提高處理系統(tǒng)的可用性。實(shí)現(xiàn)一個(gè)松耦合、集群、具有分布式并行處理特征應(yīng)用軟件系統(tǒng)的關(guān)鍵技術(shù)之一是，使用消息隊(duì)列（Message Queue，簡(jiǎn)為MQ）實(shí)現(xiàn)復(fù)雜事件處理步驟間相關(guān)數(shù)據(jù)和狀態(tài)簽名在不同處理設(shè)備之間的傳遞，以有效地減少訪問數(shù)據(jù)庫(kù)的相關(guān)入庫(kù)數(shù)據(jù)、降低I/O和避免人工干預(yù)。

傳統(tǒng)的商業(yè)智能（Business Intelligence，簡(jiǎn)為BI）技術(shù)通過數(shù)據(jù)倉(cāng)庫(kù)將數(shù)據(jù)整合到一起，并用專門的工具在數(shù)據(jù)模型的基礎(chǔ)上進(jìn)行分析，解決應(yīng)對(duì)復(fù)雜性。在技術(shù)和用途上，顯然MQ和BI是正交的。MQ不適合處理復(fù)雜性，而BI主要適應(yīng)于結(jié)構(gòu)化的歷史數(shù)據(jù)分析，無法處理“即時(shí)”的情況，CEP技術(shù)的出現(xiàn)解決了上述2個(gè)方面的問題，在實(shí)時(shí)性和復(fù)雜性方面都得到了很好的解決［8］。

事件處理的2個(gè)主要模型是數(shù)據(jù)流處理（Data Stream Processing，簡(jiǎn)為DSP）模型［8］和CEP模型［9］。Esper是純Java開發(fā)的用于CEP和DSP應(yīng)用程序的組件，可以監(jiān)測(cè)事件流，并在特定事件發(fā)生時(shí)觸發(fā)某些動(dòng)作。Esper引擎是為滿足對(duì)事件進(jìn)行分析并做出反應(yīng)等針對(duì)CEP應(yīng)用需求而產(chǎn)生的組件。CEP應(yīng)用要求實(shí)時(shí)或者接近實(shí)時(shí)處理事件（或消息）。這類應(yīng)用具有高吞吐量、低響應(yīng)時(shí)間和復(fù)雜計(jì)算等特點(diǎn)［10］。由于Esper本身不支持集群環(huán)境，無法實(shí)現(xiàn)在線生產(chǎn)系統(tǒng)的高可用性。針對(duì)這一問題，本文研究和提出使用Zoo Keeper作為管理控制臺(tái)與多個(gè)復(fù)雜事件處理引擎的交互媒介，從而擴(kuò)展復(fù)雜事件處理系統(tǒng)的集群能力，并為控制臺(tái)和處理引擎間提供一種新的交互方式。

本文設(shè)計(jì)和提出的以Esper為CEP引擎的CEP系統(tǒng)，主要由規(guī)則部署包開發(fā)環(huán)境、CEP分析引擎、運(yùn)維控制臺(tái)、ZooKeeper和數(shù)據(jù)庫(kù)服務(wù)器（Database Server，簡(jiǎn)為DBS）等組成，是一個(gè)集規(guī)則部署包開發(fā)、系統(tǒng)控制運(yùn)維管理和事件分析處理等的集成應(yīng)用系統(tǒng)。該系統(tǒng)為松耦合關(guān)系，各子系統(tǒng)只遵循相應(yīng)的約定，可以單獨(dú)實(shí)現(xiàn)并運(yùn)行。其系統(tǒng)間的交互關(guān)系圖如圖1所示。

圖1中，CEP Studio是基于開源集成開發(fā)環(huán)境Eclipse開發(fā)的插件，用于開發(fā)規(guī)則部署包。運(yùn)維控制臺(tái)和業(yè)務(wù)控制臺(tái)是基于Nutz等技術(shù)開發(fā)的Web應(yīng)用，用于集群管理、規(guī)則管理、狀態(tài)監(jiān)控等，通過控制臺(tái)配置的集群配置選項(xiàng)、規(guī)則實(shí)例參數(shù)等信息存儲(chǔ)在數(shù)據(jù)庫(kù)中。Zoo Keeper用于擴(kuò)展系統(tǒng)的集群能力，防止單點(diǎn)故障并提供分布式的同步，使系統(tǒng)具有高可靠性；其可靠性技術(shù)指標(biāo)應(yīng)能夠達(dá)到99.99%，并作為控制臺(tái)和引擎間的交互媒介。

圖1 子系統(tǒng)間的交互關(guān)系圖

3 系統(tǒng)設(shè)計(jì)與實(shí)現(xiàn)

3.1 軟件系統(tǒng)總體結(jié)構(gòu)

設(shè)計(jì)基于Zoo Keeper的軟件系統(tǒng)的總體結(jié)構(gòu)如圖2所示。

圖2 基于ZooKeeper的軟件系統(tǒng)總體結(jié)構(gòu)視圖

圖2中，控制臺(tái)與CEP引擎不直接交互，系通過Zoo Keeper間接交互，Zoo Keeper是整個(gè)系統(tǒng)的核心。只要將開發(fā)完成的規(guī)則部署包上傳到運(yùn)維管理控制臺(tái)，運(yùn)維管理控制臺(tái)就將規(guī)則模板的相關(guān)信息和規(guī)則模板存放到對(duì)應(yīng)數(shù)據(jù)庫(kù)和模板庫(kù)（文件系統(tǒng)）。業(yè)務(wù)人員通過運(yùn)維管理控制臺(tái)進(jìn)行規(guī)則實(shí)例的維護(hù)操作，每次的維護(hù)結(jié)果存儲(chǔ)在對(duì)應(yīng)數(shù)據(jù)庫(kù)和Zoo Keeper中；多個(gè)CEP引擎監(jiān)聽到Zoo Keeper中的實(shí)例出現(xiàn)變化時(shí)，會(huì)從模板庫(kù)下載規(guī)則模板到本地文件系統(tǒng)，用于維護(hù)引擎中的規(guī)則實(shí)例。

3.2 系統(tǒng)的主要設(shè)計(jì)思路

系統(tǒng)的主要設(shè)計(jì)思路是靈活運(yùn)用了Zoo Keeper的發(fā)布與訂閱特性以及發(fā)送更新通知的機(jī)制。對(duì)應(yīng)本系統(tǒng)設(shè)計(jì)，可設(shè)計(jì)為3種角色，即業(yè)務(wù)邏輯、Zoo Keeeper和規(guī)則管理。業(yè)務(wù)邏輯主要是監(jiān)聽對(duì)應(yīng)自己規(guī)則的Zoo Keeeper節(jié)點(diǎn)，1個(gè)業(yè)務(wù)邏輯對(duì)應(yīng)1個(gè)規(guī)則；Zoo Keeeper提供發(fā)布、訂閱和更新通知的功能；規(guī)則管理對(duì)應(yīng)于圖2中的運(yùn)維、業(yè)務(wù)控制臺(tái)。系統(tǒng)的業(yè)務(wù)處理流程如圖3所示。

圖3 系統(tǒng)的業(yè)務(wù)處理流程圖

業(yè)務(wù)處理流程概述如下：①業(yè)務(wù)邏輯服務(wù)監(jiān)聽對(duì)應(yīng)的Zoo Keeper節(jié)點(diǎn)；②用戶通過運(yùn)維管理控制臺(tái)更新規(guī)則；③Zoo Keeper通知各邏輯節(jié)點(diǎn)；④各個(gè)邏輯節(jié)點(diǎn)（CEP引擎）下載新規(guī)則到本地，并更新規(guī)則實(shí)例。

3.3 系統(tǒng)實(shí)現(xiàn)

系統(tǒng)實(shí)現(xiàn)的主要工作，對(duì)應(yīng)圖3的業(yè)務(wù)流程，利用ZooKeeper提供的編程接口進(jìn)行：

（1）創(chuàng)建Zoo Keeper節(jié)點(diǎn)；

（2）設(shè)置Watcher回調(diào)方法，在本系統(tǒng)中為下載規(guī)則的名稱、參數(shù)名稱、狀態(tài)和操作結(jié)果等，從而使得各個(gè)CEP引擎能夠同步更新本地的規(guī)則實(shí)例。

相關(guān)實(shí)現(xiàn)的關(guān)鍵代碼如下：

∥Watcher實(shí)例，回調(diào)方法為下載規(guī)則；

Watcher wh=new Watcher（）﹛

public void process（WatchedEvent event）﹛

download rule；

﹜；

∥創(chuàng)建一個(gè)Zoo Keeper實(shí)例，第一個(gè)參數(shù)為目標(biāo)服務(wù)器地址和端口，第二個(gè)參數(shù)為Session超時(shí)時(shí)間，第三個(gè)參數(shù)為節(jié)點(diǎn)變化時(shí)的回調(diào)方法；

ZooKeeper zk=new ZooKeeper（"10.15. 82.166：3351"，500000，wh）；

∥創(chuàng)建一個(gè)節(jié)點(diǎn)root，數(shù)據(jù)是mydata，不進(jìn)行ACL權(quán)限控制，節(jié)點(diǎn)為永久性的（即客戶端shutdown了也不會(huì)消失）

zk.exists（"/root"，true）；

zk.create（"/root"，"mydata".getBytes（），Ids.OPEN_ACL_UNSAFE，CreateMode. PERSISTENT）；

3.4 仿真試驗(yàn)示例及其分析

限于時(shí)間和資源的限制，本文采用3臺(tái)PC機(jī)進(jìn)行的仿真試驗(yàn)示例及其分析簡(jiǎn)潔描述如下。

3.4.1 ZooKeeper的集群配置

用3臺(tái)機(jī)器構(gòu)成ZooKeeper集群，具體機(jī)器名和IP地址對(duì)應(yīng)如下：

ZK01 192.168.0.1

ZK02 192.168.0.2

ZK03 192.168.0.3

Zoo Keeper的版本為3.3.4。以ZK01機(jī)器為例（ZK02和ZK03的配置文件與ZK01完全相同），具體相關(guān)配置示例說明如下：

（1）/etc/hosts

ZK01 192.168.0.1

ZK02 192.168.0.2

ZK03 192.168.0.3

（2）ZooKeeper配置文件

tick Time=2000

dataDir=/home/hadoop/storage/zookeeper

clientPort=2181

init Limit=5

sync Limit=2

server.1=ZK01：2888：3888

server.2=ZK02：2888：3888

server.3=ZK03：2888：3888

3.4.2 Esper處理規(guī)則的測(cè)試用例

以統(tǒng)計(jì)軌道交通5 min內(nèi)乘客出站扣款票價(jià)等于某幣值的總?cè)藬?shù)為例，可表示為：

EPL：count*from transaction_exit.win. time（5 minute） where ticket_price=400；

EPL：count*from transaction_exit.win. time（5 minute） where ticket_price=500；

price=400和price=500分別表示幣值單位為“分”的出站扣款，在這里僅為表示復(fù)雜事件處理中的某種規(guī)則變化示例。

3.4.3 系統(tǒng)的可靠性和實(shí)時(shí)性

由于Zoo Keeper采用了分布式集群的方式，所以保證了系統(tǒng)可靠性（Zoo Keeper集群中的服務(wù)器越多，其可靠性越高）。多個(gè)ZooKeeper采用選舉的方式選出一個(gè)leader，同時(shí)Zoo Keeper采用了事件通知方式，所以系統(tǒng)的實(shí)時(shí)性也有了較大的提高，基本上達(dá)到了完全同步的狀態(tài)。

4 結(jié)語

本文提出和完整描述了基于Zoo Keeper且以Esper為核心的復(fù)雜事件處理系統(tǒng)的一種技術(shù)解決方案，包括實(shí)現(xiàn)該技術(shù)方案的系統(tǒng)軟件構(gòu)架、系統(tǒng)總體結(jié)構(gòu)、業(yè)務(wù)處理流程、系統(tǒng)實(shí)現(xiàn)和仿真試驗(yàn)示例及其分析。

結(jié)合ZooKeeper的適合分布式計(jì)算集群協(xié)調(diào)及管理的高可用性能力和Esper適用于CEP、DSP應(yīng)用組件的優(yōu)勢(shì)，提出了一種適合CEP和DSP應(yīng)用的高可用性的技術(shù)解決方案，目前仍鮮見。

由于受到時(shí)間、資源和其它因素的限制，本文對(duì)系統(tǒng)的實(shí)現(xiàn)方法僅在3臺(tái)PC機(jī)上進(jìn)行了仿真驗(yàn)證。仿真結(jié)果證明本文的研究方法和給出的技術(shù)解決方案是可行的。

［1］ 湯玲麗.復(fù)雜事件處理引擎關(guān)鍵技術(shù)研究［D］.哈爾濱：哈爾濱工程大學(xué)，2012.

［2］ 史喜陽，孫棣華，宋瀟瀟.基于CEP的RFID數(shù)據(jù)處理模型研究［J］.自動(dòng)化技術(shù)與應(yīng)用，2008，27（4）：73.

［3］ 劉卓楊.復(fù)雜事件處理中時(shí)間窗口的改進(jìn)［J］.計(jì)算機(jī)工程，2010（10）：50.

［4］ 李汝光，趙俊.基于ZooKeeper的分布式緩存的設(shè)計(jì)與實(shí)現(xiàn)［J］.綿陽師范學(xué)院學(xué)報(bào)，2011，30（11）：116.

［5］ Robinson H，王旭.Observer：讓ZooKeeper更具可伸縮性［J］.程序員，2010（2）：91.

［6］ Tan Xuehou.Shortest zookeeper's roles in simple polygons［J］.Information Processing Letters，2001（1）：25.

［7］ 趙時(shí)旻.軌道交通自動(dòng)售檢票系統(tǒng)［M］.上海：同濟(jì)大學(xué)出版社，2007.

［8］ 蔡昭權(quán)，索劍，汪華斌，等.基于Esper和Nagios的網(wǎng)絡(luò)監(jiān)控系統(tǒng)設(shè)計(jì)與實(shí)現(xiàn)［J］.計(jì)算機(jī)工程與科學(xué)，2012，34（9）：8.

Application of ZooKeeper in Complex Event Processing System

Feng Juan

Distributed calculation processing and coordinating of the complex event processing has been a difficult problem for distributed calculation，especially for a high availability technical practice based on low cost.A technical solution is proposed，in which Zoo Keeper is put into complex event processing application in terms of the core of Esper.This solution can coordinate multiple complex event processing engines and operate as an interactive medium among control consoles and complex event processing engines，realize a cluster technical practice for distributed calculating of complex event processing，raise system availability and real-time performance.The solution proposed in this paper has been validated by the construction of a simulation experiment platform of 3 PCs，and the result proves to be practical.

AFCsystem；device monitor；data stream processing；distributed calculation；cluster technology

TP 274

2014-06-17）