主動式復雜事件處理方法的研究

耿少峰，王永恒，李仁發，張佳

（1. 湖南大學信息科學與工程學院，湖南 長沙 410082；2. 集美大學計算機工程學院，福建 廈門 361021）

主動式復雜事件處理方法的研究

耿少峰1,2，王永恒1，李仁發1，張佳2

（1. 湖南大學信息科學與工程學院，湖南 長沙 410082；2. 集美大學計算機工程學院，福建 廈門 361021）

在CPS的應用背景下，對傳感器和控制設備產生的不確定性事件流進行分析和處理得出高層事件，然后在此基礎上引入適應性動態貝葉斯網絡和并行馬爾可夫決策過程模型來支持主動式的復雜事件處理。針對大型CPS中馬爾可夫決策過程存在的狀態數量巨大的問題，引入狀態劃分和報酬分解的方法來進行并行優化。在模擬交通網絡的環境中，實驗結果顯示所提方法能有效地處理事件流，并具有良好的可擴展性。

信息物理融合系統；大數據；復雜事件處理；馬爾可夫決策過程

1 引言

信息物理融合系統（CPS, cyber physical system）將智能計算、通信和控制技術深度融入到物理系統，使計算資源與物理資源緊密結合與協調分配，完成信息世界與物理世界的融合。它的應用領域非常廣泛，小到智能家庭網絡，大到工業控制系統，覆蓋了各種不同規模的應用系統。如何有效地處理 CPS應用中所產生的大數據是人們目前需要面對的主要問題。當前在大數據研究領域，對于歷史數據的分析和處理已經進行了廣泛深入的研究，特別是在數據倉庫和數據挖掘方面。常用方法之一是將大量的服務器組成集群或使用專業的云計算平臺（如亞馬遜的AWS）以及基于Hadoop的分布式計算框架。另外一種近期常見的計算平臺是Spark，它突破了 MapReduce的算法框架，支持更靈活的數據處理方式，性能也有了很大的提升。但這些方式并不適合于 CPS的數據處理。因為 CPS在工作時產生的是實時的數據流，對這些數據流的處理往往有較高要求的時間約束。而 Hadoop/MapReduce的計算框架僅適用于處理靜態數據和歷史數據。近期也出現了一些面向數據流的處理平臺，如Spark streaming、Storm和Flink。這些平臺能夠高效地處理數據流，但僅限于比較簡單的過濾、聚合等操作，無法直接支持面向CPS的復雜數據流處理。

在 CPS應用中，系統必須處理來自無線傳感網、RFID閱讀器、GPS、社會媒體等各種各樣設備的信號，這些信號可以被當成事件來處理。這些由設備直接生成的事件稱為原始事件。一般情況下原始事件里的語義信息非常有限，在實際應用中，人們更加關注的是業務邏輯和規則等高級別信息。如在裝有智能門禁的小區門口有車輛通行時，RFID閱讀器會進行讀操作來生成原始的事件，但只有“車輛進入或離開小區”這樣的復雜事件才是人們真正關心的，可以根據一些模式將許多原始事件結合在一起，從而得到對用戶有意義的復雜事件。CPS應用系統先將業務邏輯轉換成復雜事件，再通過處理復雜事件來檢測業務邏輯。復雜事件處理(CEP,complex event processing)[1]用于處理數量巨大的簡單事件，并從中得到有價值的高層次的信息。

在某些CPS的應用中，對象執行的操作可以改變系統的狀態。當前主流的事件處理方法為響應式的，這意味著動作是由系統狀態發生變化觸發。主動式事件處理系統通過運用預測和自動決策技術[2]減輕或避免將來不希望的事件發生。例如，在一個CPS的交通網絡中，可以根據當前的狀態和歷史數據來預測道路的擁擠狀態，然后采取一些行動以避免將來可能出現的擁堵。馬爾可夫決策過程（MDP,Markov decision process）作為一個最佳決策過程中的隨機動態系統，多年來人們已對其進行了深入的研究，并在多個領域中得到了廣泛的應用。

目前，基于MDP實現主動式復雜事件處理還面臨著巨大的挑戰。首先是如何有效地處理由于誤讀、干擾等原因產生的事件的不確定性。當前的研究主要限于通過信息融合和推理來解決單個事件的不確定性，但缺乏對復雜事件不確定性的研究。另一個問題是如何進行高效準確的狀態預測。盡管人們已經對數據流預測進行了大量的研究，但針對CPS產生的高速、海量數據流，預測模型和算法上都還面臨挑戰。同時，在決策支持方面，目前很少有文獻涉及如何將MDP融入到主動式復雜事件處理中，以及如何應對在此過程中MDP所產生的巨大的狀態空間。

針對上述問題，本文提出一種主動式復雜事件處理的架構和方法。該方法以交通網絡為背景，將樹的匹配和緩沖鏈表相結合，高效地處理了大量由車輛產生的不確定事件流，然后將處理后的復雜事件提交給適應性動態貝葉斯網絡（ADBN, adaptive dynamic Bayesian network）來預測未來事件和系統狀態，最后主動式響應器根據 ADBN的預測結果，利用一種基于狀態劃分和并發操作的新型并行MDP方法進行分析，最終產生避免交通擁堵的決策。

2 相關研究

CEP通過監測不斷產生的數據流，根據每個事件發生的集合/序列識別復雜事件，然后對檢測到的狀態做出回應。Etzion等[3]在其著作中定義了復雜事件處理的基本概念及體系結構。一個事件處理agent（EPA, event processing agent）執行某些處理邏輯，從輸入事件中提取出復雜事件。多個EPA連接在一起可以執行更復雜的層次化事件處理，這種網絡稱為事件處理網絡。

在以前的主動數據庫研究中，人們已經對 CEP做了大量的工作。常用的方法都是基于固定的數據結構，如樹、圖、自動機和 Petri網。這些方法很難靈活地對事件查詢語言的實現進行優化，也很難根據應用需求的變化來支持查詢語言的擴展。針對這些問題，Eugene等[4]提出了一種基于查詢規劃的復合事件檢測方法（SASE）。SASE基于有限自動機（NFA）和堆棧來檢測復雜事件，并對大的滑動窗口和大的中間結果集進行了優化，實現了比較好的性能和可擴展性。近年來的很多工作圍繞著對SASE的改進展開，AGRAWAL等[5]提出了改進的有限狀態機模型來提供更豐富的查詢能力，Zhang等[6]對SASE進行了改進，提供了對非精確時標的事件流的支持。

由于傳感器噪聲等因素，CPS事件具有不確定性，這種不確定性通常使用概率來表示。處理概率性原始數據的CEP普遍采用的模型有隱馬爾可夫模型、動態貝葉斯網絡和條件隨機域等。近年來也出現了一些基于有限狀態機的概率復雜事件處理方法。文獻[7]基于NFA設計了一種新的數據結構——鏈接實例隊列（chain instance queues），并采用一種條件概率索引樹來提高搜索性能。另外，文獻[8]面向NFA提出一種優化方法，不僅能夠計算復雜事件的概率，還能夠對不可信網絡產生的數據獲取復雜模式的可信度。基于有限狀態機的概率復雜事件處理，比較有代表性的研究項目是美國華盛頓大學的Cascadia系統[9]和Laha系統[10]。Cascadia系統基于概率數據庫中的可能世界模型（possible world model），實現了對概率事件流和歷史概率事件的查詢，但沒有在預處理中考慮流上相鄰事件間的時間關聯性。Laha成功支持了關聯概率事件流上的高效查詢，并支持更復雜的事件代數操作，從而能支持更靈活的復雜事件處理。

主動式系統的研究已經持續了多年，如主動式運輸管理系統[11]、面向即時無線網絡的主動型路由[12]等。這些研究總體上代表了一種由響應式系統向主動式系統轉變的潮流。在復雜事件處理這個領域，對主動式方法的研究還比較少。但當前CPS的發展已經對這種研究提出了需求，也提供了研究的基礎條件。Engel等[13,14]提出了一種基于事件的主動式計算框架，該框架主要由復雜事件處理網絡、預測器和基于馬爾可夫決策過程的主動式響應器構成。該工作對傳統的復雜事件處理agent進行了擴展，支持2種新型的agent：預測agent用于預測系統狀態；主動式響應agent用于計算合適的響應并執行。但它的應用背景不是面向CPS，也沒有考慮大規模的數據和狀態的問題。

多年來，人們對MDP進行了廣泛的研究和應用，產生了多個變種。對于大型CPS應用這個領域，最大的挑戰在于巨大的狀態空間造成的組合爆炸問題。對這個問題當前主要有2個研究方向。第一類是根據問題領域中的信息來簡化結構，如狀態聚集的方法[15]和時間聚集的方法[16]等。此類方法的問題和具體的系統相關，無法找到普遍的解決方法。第二類是近似計算的方法，如近似動態規劃[17]和選擇性近似[18]等方法。這類方法的關鍵是如何在優化過程中對值函數進行近似。當應用到大型CPS的事件處理領域，MDP的規模更大且具備一些新的特征，其具體模型和大規模計算問題的解決仍然是一個挑戰。

圖1 系統架構

3 系統架構

本文方法的總體結構如圖1所示。CPS中各類傳感器及RFID等設備產生的信號，匯聚在一起稱為原始事件流。由于設備本身的精度及干擾等原因，這些事件不是完全精確的，稱為概率事件。這些原始事件經過概率復雜事件處理引擎的處理，提取出有意義的高層事件，也就是復雜事件。其中，PEPA（probabilistic event processing agent）為概率事件處理器[19]，它負責處理由底層設備產生的原始事件流，并最終生成概率復雜事件。查詢規劃器根據復雜事件查詢語句使用一個或多個 PEPA來完成查詢。多個 PEPA組合在一起構成的網絡稱為概率事件處理網絡（PEPN, probabilistic event processing network）。本系統中預測分析器（PA, predictive analytics）的實現方法是基于Pascale等[20]的工作，通過使用一種適應性動態貝葉斯網絡模型，根據輸入的復雜事件來預測道路網絡中可能會發生擁堵的節點。復雜事件被存儲在事件數據庫中，PA可以通過學習這些歷史事件獲得更準確的預測。主動式響應器根據 PA預測出來的系統未來狀態，利用并行MDP產生避免或減弱這些狀態發生的決策。PRA（proactive agent）為執行器，負責執行對應的決策動作。上下文可以從復雜事件中獲取，也可以從其他信息系統獲取，并影響系統的狀態預測和控制。

定義1 概率原始事件CPS事件流中的原始事件即在某個時間的一次發生（一次信號讀?。?。概率原始事件表示為：lt;ID, A, T, Prgt;。其中，ID為產生數據的設備標簽號，A為事件屬性的集合，T為事件發生的時標。Pr為事件發生的概率值，它代表由原始信號轉換為確切的CPS事件的概率。

定義 2 概率復雜事件是由原始事件或者復雜事件按照一定的運算規則形成的事件。一個概率復雜事件表示為：lt;E, R, Tc, Prgt;。其中，E是符合合成規則的概率原始事件串，R為事件合成規則，Tc為事件的時間跨度，Pr為事件概率值。

事件類型由一組具有相同語義和結構的事件規則構成，每個事件對象都是一個事件類型的實例。一個事件類型可以表示為原始或復雜事件。主要的復雜事件模式包括 ALL、ANY、COUNT和SEQ[3]等。如在交通系統中，COUNT模式用來計算規定時間內某一地點經過車輛的數量。把這些模式結合起來可以構建層次性的復雜模式。

4 主動式復雜事件處理

4.1 不確定性事件流的處理方法

由于RFID閱讀器等設備存在漏讀、誤讀和臟讀等情況，所以導致交給PEPA處理的數據存在不確定的可能性。這些不確定的原始數據被稱為不確定事件。不確定事件流的復雜事件處理主要有2個挑戰，一是如何對大量高速的、實時的事件進行符合查詢規則的檢測；二是如何計算由相關或不相關的不確定事件流組成的復雜事件的概率。本文提出了一種面向不確定性事件流的復雜事件處理方法(ISCEP, indeterminate stream complex event processing)來解決上述問題。該方法采用事件概率模型建模，并通過基于匹配樹與各節點的緩沖鏈表相結合的方法來處理大量的事件流。

在大規模的 CPS應用中經常需要處理分布式的事件流，本文假設不同事件流上的事件實例是相互獨立的。在單一事件流里，一些在SEQ模式中的原始事件具有馬爾可夫性。這意味著下一個事件發生的概率僅取決于序列中的當前事件，和歷史事件無關。如一輛汽車在i+1時刻的位置和它在i時刻的位置有關，而和i時刻之前的位置無關。一個具有馬爾可夫性的事件序列被稱為馬爾可夫鏈。像Kawashima等[9]做的工作那樣，使用條件概率表（CPT）來處理條件概率。除了這些馬爾可夫鏈的事件，可能還存在一些相互獨立的原始事件。對于一個SEQ事件E=SEQ(e1, e2,…,en)，原始事件劃分為2個集合S和T。S包含相互獨立的事件，T包含一個或多個馬爾可夫鏈。SEQ事件的概率可以根據式(1)計算。

其中，Si是非獨立集合 S中的某一條馬爾可夫鏈，e1表示馬爾可夫鏈中的第一個事件，Pr(em+1|em)表示連續發生的事件是前后相關的，它的值可以在CPT中獲得。

事件概率模型中PEPA的輸入是由概率原始事件組成的事件流。閱讀器所處區域用大寫字母表示，時標用數字表示。如圖2所示的閱讀器在不同區域和時刻讀取到由同一個設備產生的概率流。

圖2 概率原始事件流

EPA的輸出是概率復雜事件。例如表1是經過處理后得到的某個車輛可能的行駛路徑，其中省略了復雜事件的合成規則和時間跨度，保留了2個主要的屬性：符合合成規則的概率原始事件串和復雜事件的概率。根據EPA的輸出，PA可以選擇更可靠的結果作為輸入，這樣在一定程度上可以避免因數據錯誤而導致錯誤的狀態預測。

表1 概率復雜事件輸出流

不確定性事件流的具體處理流程如圖3所示。

圖3 不確定性事件流處理方法

當查詢被提交的時候，會注冊相應的復雜事件。所謂注冊就是在查詢映射表中建立查詢索引，同時建立用于流處理的匹配樹。將交通網絡劃分為若干子區域，根據子區域和不同的時間段（如早晚高峰期）對應的已注冊查詢來構建查詢映射表。當某一時段內某個區域發生事件的時候，就可以通過查詢映射表迅速地找到已注冊查詢。如構造一個如表 2所示的查詢映射表。如果從輸入流中讀取到8:00區域A在發生事件，那么應該將原始事件發送到查詢1對應的匹配樹處理。

表2 查詢映射表

當處理實時數據流時，每個登記的查詢都會建立一個匹配樹。基本的建立規則在文獻[21]描述。葉子節點是區域節點，中間節點是復合規則節點。在每一個節點的下方都會建立一個緩沖鏈表用以緩沖歷史事件。如圖4所示是查詢1的匹配樹構造結果，其中，SEQ節點是由B、C區域事件節點組成，TSEQ節點是由復合節點SEQ和A區域事件節點組成。

在執行查詢時，首先獲得概率簡單事件流中的每一個區域事件的相關查詢集合。接著遍歷每一個查詢，把區域事件輸送到查詢對應匹配樹中，按照自底向上的方式進行匹配。如果當前節點匹配成功就會在本節點的緩沖鏈表中加入匹配成功的記錄，接著發送更新消息給父節點。父節點接到更新消息就會對所有子節點的緩沖鏈表進行匹配，若滿足規則會在自己的緩沖鏈表中加入成功匹配的節點記錄，然后繼續向上級節點發送更新消息。如此迭代直到沒有節點再接收到更新消息為止。最后對匹配樹的根節點緩沖鏈表進行查看，如果有成功匹配記錄，則自頂向下輸出所有過程并返回結果集合，否則直接結束。具體過程如算法1所示。

圖4 查詢1的匹配樹

算法1 ISCEP算法

輸入 E為概率原始事件流，QTree為查詢匹配樹

輸出 R為概率復雜事件模式的實例集合

方法：

算法中幾個函數含義如下。

Add()：事件匹配成功后，將記錄加入鏈表。記錄包含新計算的復雜事件概率、產生此記錄的事件集合和時標。

Update()：節點緩沖鏈表有新記錄加入后，用此函數向節點的父節點發送更新消息。

Match()：對子節點緩沖隊列中的記錄按照時標順序由后往前進行遍歷，對時標限制和運算符限制進行匹配。

Output(QTree)：根節點緩沖鏈表有記錄時候調用。因為成功匹配記錄包含直接子節點的信息，所以按照自頂向下的方式檢索直到葉子節點，就能得到組成當前查詢結果的所有信息。

設已注冊的查詢有N個，表示為Q1,…,QN，劃分的區域有M個，表示為D=D1,…,DM，函數f(Di)表示區域Di對應的查詢的集合。設一個新的查詢Q對應K個區域，表示為DQ=DQ1,…,DQK，其中，DQi∈D。在沒有優化的情況下，Q的查詢代價表示為，QDi表示對應到區域 DQi上的子查詢。如果某個區域DQi上的子查詢和已注冊查詢匹配，則其代價可表示為find(f(DQi))+ merge(f(DQi))，其中，merge(f(DQi))為合并子查詢結果的代價，而由于建立了索引，find(f(DQi))可忽略不計。設已注冊查詢匹配率為 α，則查詢 Q的區域匹配數為對應的區域集合為D'Q。則Q的查詢代價變為。由于merge合并結果的代價遠小于直接查詢，因此當匹配率α的值比較大的時候，算法的查詢性能會有大幅度的提高。

4.2 基于并行馬爾可夫決策過程的決策方法

針對CPS復雜事件處理系統的特點，傳統馬爾可夫決策過程擴展如下。

定義 3 主動式復雜事件系統并發馬爾可夫決策過程。一個主動式復雜事件系統并發馬爾可夫決策過程定義為。其中，I為處理動作的agent有限集合，S為狀態的有限集合，包含一個特殊的初始狀態為動作聯合，其中，Ai為來自第 i個 agent的動作。P: S×A×S→[0, 1]為狀態轉換函數，P(s,α,s')代表在狀態 s執行響應 α后轉換到狀態 s'的概率，R:S→Re為報酬函數（Re為實數）。對每個為CPS的復雜事件，復雜事件的變化影響系統的狀態。C為上下文，在不同的上下文中，agent會采用不同的動作為復雜事件對系統狀態的影響。

系統從原始狀態開始，根據合適的策略選擇一個或多個動作，由動作agent執行。動作agent的操作引起 CPS原始事件的變化，原始事件的變化引起復雜事件的變化，根據對復雜事件的綜合分析，確定系統轉移到下一個狀態的概率。馬爾可夫決策過程的關鍵是尋找一個策略π :S→A，它把每個狀態映射到一個動作集合。最優的策略使報酬的總和達到最大。策略選擇步驟基于式(2)。

策略更新步驟基于式(3)。

其中，γlt;1為衰減因子。經典的MDP中動作是順序執行的，最近有些研究提到并發MDP按功能對不同的動作進行分組，這些分組后的動作并發執行且相互沒有影響。MDP模型針對一個狀態有多個并發執行的動作，它們在不同的位置執行相同的功能來協同把系統轉向期望的狀態。本文中的MDP方法基于以下觀察結果。

1) 觀察結果1。大型CPS中的對象可以根據上下文進行劃分，每個組對應一個子狀態，而不單獨考慮每個對象的狀態。

例如在考慮交通擁堵問題時，每個路口的車流量（或擁堵情況）是子狀態，所有路口的擁堵狀況構成系統的總體狀態。

2) 觀察結果2。大型CPS中某個位置對應的子狀態依賴于其鄰居位置的狀態（此處的鄰居是指在一定范圍內和當前節點相連的其他節點）。

例如某個路口的擁堵狀態依賴于前一個時間段其鄰居路口的擁堵狀態。

3) 觀察結果3。在大型CPS中，可以把狀態節點分為一般節點和重點節點。重點節點是在主動式處理中要控制其狀態的節點。如圖5所示，可以根據重點節點對交通網絡進行劃分。對單個子網中重點節點的狀態轉換，只需調整其子網內鄰居節點的狀態來解決。

交通網絡可以如圖5劃分成若干組。一個或多個擁堵程度高（由PA預測）的節點作為一個組的中心，所有到中心的距離小于閾值D的節點劃分為一個組。假設有 n個節點：J={j1,j2,…,jn}，本文把交通流量按擁堵程度分成 k個級別：CS={c1,c2,…,ck}。MDP中的狀態可以表示為st=lt;st1,st2,…, stngt;，其中，sti∈S 是節點 ji在 t時刻處于擁堵狀態，si∈CS。策略πit={ait1,ait2,…,aitn}，其中，動作aitk的作用是在t時刻引導一些車輛改變它們的路徑（如可以通過導航軟件提示等方法），從而減少k組的擁堵程度。本文假設在交通網絡中分組時，任意 2組中心間的距離足夠大，這意味著子動作間不存在依賴。因此上述分組的數據可以用分布式服務器處理，通過并行來提高系統性能。通過網絡劃分，把一個狀態數為n的MDP轉換成多個子MDP，每個子MDP的聯合狀態數遠小于n，從而大大降低了問題規模。

圖5 基于重點節點的網絡劃分

算法2 并行調度方案

輸入 已劃分的交通網絡分組V={v1,v2,…,vk}，分布式系統中服務器節點C={c1,c2,…,cn}

輸出 網絡分組在處理節點中的分配方案

方法：

其中，分布式系統為同構體系結構，并且服務器節點數目n小于子網分組數k，vk.s是分組vk中所有節點擁堵程度總和。根據vk.s的值對V進行排序，放入堆 MaxHeap中。遍歷 MaxHeap，通過Assign(cn,vk)將當前分組任務分配給 cost最小的服務器，隨后對服務器的cost值進行修正。此算法實現了分布式服務器的負載均衡，相對于單節點，處理性能會有大幅度的提高。

根據觀察結果，在前述MDP模型中加入新的變量G，代表子狀態節點的網絡結構。G=(V,E)，其中，V為節點的集合，E為邊的集合。一個邊(i,j)存在表示節點i影響節點j的狀態。

定義4 鄰居狀態節點。對于任意狀態節點i∈V，其鄰居狀態節點N(i)={j∈V |di,j≤k}，其中，di,j為i和j之間的路徑長度，k是閾值。定義4表明一個節點的狀態只受鄰居節點狀態的影響。

通過定義鄰居節點，和重點節點相關的聯合狀態數進一步縮小，從而進一步降低問題的規模。

定義5 狀態轉換分解。設有n個狀態的子節點，對應的動作也分為n個子動作，則根據前面的觀察結果，有

其中，s表示當前狀態，s'表示下一個狀態。

定義6 報酬分解。設有n個狀態的子節點，對應的動作也分為n個子動作，則根據前面的假設，有

定義6意味著整體報酬被分解為若干個組的報酬。每一組的報酬又進一步分解為所有鄰居節點的報酬。這樣每個節點的報酬可以采用單機多處理器的并行方式來計算。

在交通CPS中，當一個動作ai被執行，節點i處的交通流量為

其中，Pa_out(ji)是嘗試減少節點i處流量的節點集，Pa_in(ji)是嘗試增加節點 i處流量的節點集，αki是Sk流向節點i的比例。假設并不是所有車輛都會按本文的策略來改變行駛路線，那么 βki是 Sk受動作ai影響的比例，P(jk)是在節點k處的車輛根據ai改變路線的概率，它可以從歷史數據得到。整體的狀態事務概率可以用P(j)和?i計算。

根據定義 6，可以定義平均流量和其中一組gi的方差為

本文的目標是把所有擁堵狀態最小化。因此，定義歐式距離子動作的報酬函數和整體報酬函數為

最終，最優策略可以按式(10)計算。

5 實驗結果與分析

SUMO是一個開源的微觀道路交通仿真模擬器[22]，它可以檢測通過相應區域車輛的數量和每輛車的位置。如圖6所示，本文用JADE(Java agent development framework)框架對 SUMO進行擴展來支持多agent系統，同時使用 SUMO的TraCI接口來獲得感應圈變量和車輛位置的變量，用它們來模擬RFID和GPS閱讀器?；谶@個框架，agent控制器用智能算法控制車輛在 SUMO中移動，從中獲取事件數據和執行相應的動作。在實驗中，在地圖里設置了80個路口和8萬輛汽車。為了更接近真實的交通系統，定義了一組規則：每輛車都有一個家的位置和辦公室位置，車輛Vi在家和辦公室之間運行的概率是 pi。這些車輛還有相應概率去超市、醫院等其他地點。此實驗模型可以滿足 PCEP對交通網絡的劃分、狀態和動作的分解等操作，并且可以并行的生成決策方法。本文把4臺至強E3處理器和16 GB內存的服務器當作數據處理服務器。另一臺有4 GB內存的PC運行SUMO。

圖6 系統實現框架

首先來評估ISCEP關于多查詢的處理性能。本文對經典的RCEDA方法[23]進行了類似本文提出的概率計算方式的拓展，稱之為 PRCEDA(probability RCEDA)。通過PRCEDA與本文提出的ISCEP進行比較。假定所有的注冊查詢都是二叉樹且Query Depth不超過4層，從圖7中可以觀察到，隨著注冊查詢數目的加大，ISCEP的時間消耗增長幅度相比 PRECDA要小很多。因為PRECDA以輪詢方式查詢匹配樹，隨著注冊查詢數目的增加，查詢效率會越來越低。使用查詢映射表的ISCEP輪詢次數要少很多，所以查詢效率下降較少。

圖7 查詢處理性能

下一個實驗評估該系統的平均擁堵度，結果如圖8所示。本文把擁堵程度劃分為11個等級，其中，“0”表示無擁堵，“10”表示擁堵程度最高。從圖中可以觀察到使用本文方法比讓汽車自主行駛時平均擁堵度明顯降低。原因是本文方法發現節點可能出現擁堵時，會主動引導后續車輛向輕負荷節點行駛。同時，當車輛總體擁堵程度比較高的時候，本文方法帶來的擁堵下降也更多。原因是當總體車流量比較大時，系統執行的動作會影響更多的車輛，帶來更多的擁堵下降。

圖8 擁堵程度比較

最后的實驗中采用不同數量的服務器和不同大小的數據來測試本文方法的性能。重點節點的數量固定在20，其結果如圖9所示。結果表明，本文方法通過網絡的劃分、鄰居節點定義及狀態和動作分解，可以在較大規模的MDP問題中在可接受的時間內形成決策。從圖中可以看到，生成決策的平均消耗時間隨著車輛數量的增加成正比。原因在于車輛數量的增加需要執行更多的子動作，因此增加了算法的復雜性。通過本文方法還可以發現隨著車輛數量的增加，4個服務器的性能更好，而且生成決策所需的時間增長相對較慢。這說明系統具備更好的可擴展性，在一定范圍內可以通過增加新的計算資源來支持更大規模的交通網絡。

圖9 不同服務器數量的性能

6 結束語

本文提出了一種對 CPS設備產生的不確定性事件流進行處理分析，然后在基于ADBN的預測結果上引入并行MDP模型來控制大規模交通擁堵的主動式復雜事件處理方法。實驗評估表明，該方法能夠有效地減少節點的擁堵程度，并且具有良好的可擴展性。但在并行MDP中，子狀態和子響應的空間隨著節點和車輛的增加依然會變得巨大。今后的工作重點在于研究如何進一步減少子狀態的空間數量，從而提高系統的整體性能。

[1] LUCKHAM D C. The power of events: an introduction to complex event processing in distributed enterprise systems[M]. Boston, Addison Wesley, 2002.

[2] ENGEL Y, ETZION O. Towards proactive event-driven computing[C]//The Fifth ACM International Conference on Distributed Event-Based Systems. 2011: 125-136.

[3] ETZION O, NIBLETT P. Event processing in action[M]. Manning Publications, 2010.

[4] WU E, DIAO Y, RIZVI S. High-performance complex event processing over streams[C]//2006 ACM SIGMOD International Conference on Management of Data. 2006: 27-29.

[5] AGRAWAL J, DIAO Y, GYLLSTROM D. Efficient pattern matching over event streams[C]//SIGMOD Conference. 2008: 147-160.

[6] ZHANG H, DIAO Y, IMMERMAN N. Recognizing patterns in streams with imprecise timestamps[J]. PVLDB, 2010, 3(1): 244-255.

[7] XU C, LIN S, LEI W. Complex event detection in probabilistic stream[C]//The 12th International Asia-Pacific Web Conference. 2010:361-363.

[8] KAWASHIMA H, KITAGAWA H, LI X. Complex event processing over uncertain data streams[C]//The Fifth International Conference on P2P, Parallel, Grid, Cloud and Internet Computing. 2010: 521-526.

[9] KAWASHIMA H, KITAGAWA H, LI X. Complex event processing over uncertain data streams[C]//3PGCIC. 2010: 521-526.

[10] WELBOURNE E, KHOUSSAINOVA N, LETCHNER J, et a1. Cascadia: a system for specifying, detecting, and managing RFID events[C]//The 6th International Conference on Mobile Systems, Applications, and Services. 2008: 281-294.

[11] DOLEV S, KOPEETSKY M, SHAMIR A. RFID authentication efficient proactive information security within computational security[J].Theory of Computing Systems, 2011, 48(1):132-149.

[12] KUNZ T, ALHALIMI R. Energy-efficient proactive routing in MANET: energy metrics accuracy[J]. Ad Hoc Networks, 2010, 8(7):755-766.

[13] ENGEL Y, ETZION O, FELDMAN Z. A basic model for proactive event-driven computing[C]//The 6th ACM International Conference on Distributed Event-Based Systems (DEBS'12). 2012: 107-118.

[14] ENGEL Y, ETZION O. Towards proactive event-driven computing[C]//The Fifth ACM International Conference on Distributed Event- Based Systems, DEBS. 2011: 125-136.

[15] JIA Q S. On state aggregation to approximate complex value functions in large-scale Markov decision processes[J]. IEEE Transactions on Automatic Control, 2011, 56(2):333-344.

[16] SUN T, ZHAO Q C, LUH P B. Incremental value iteration for time aggregated Markov decision processes[J]. IEEE Transactions on Automatic Control, 2007, 52(11):2177-2182.

[17] POWELL W B. Approximate dynamic programming: solving the curse of dimensionality[M]. New York, Wiley-Inter Science, 2007.

[18] JIA Q S, ZHAO Q C. Strategy optimization for controlled Markov process with descriptive complexity constraint[J]. Science in China,2009, 52(11):1993-2005.

[19] WANG Y H, CAO K N, ZHANG X M. Complex event processing over distributed probabilistic event streams computers and mathematics with applications[J]. Computers amp; Mathematics with Applications,2012, 66(10): 1808-1821.

[20] PASCALE A, NICOLI M. Adaptive Bayesian network for traffic flow prediction[C]//Statistical Signal Processing Workshop (SSP). IEEE,2011: 177-180.

[21] 王永恒, 楊圣洪. 分布式 RFID數據流的復合事件檢測方法[J]. 計算機應用研究, 2011, 28(11): 4177-4179.WANG Y H, YANG S H. Complex event detection method for distributed RFID systems[J]. Application Research of Computer, 2011,28(11): 4177-4179.

[22] BEHRISCH M, BIEKER L, ERDMANN J, et al. Sumo - simulation of urban mobility: an overview[C]//The Third International Conference on Advances in System Simulation. Barcelona, Spain, 2011: 63-68.

[23] WANG F, LIU S, LIU P, et al. Bridging physical and virtual worlds:complex event processing for RFID data streams[C]//The 10th International Conference on Extending Database Technology (EDBT'2006).2006: 588-607.

Research of proactive complex event processing method

GENG Shao-feng1,2, WANG Yong-heng1, LI Ren-fa1, ZHANG Jia2
(1. College of Information Science and Engineering, Hunan University, Changsha 410082, China;2. College of Computer Engineering, Jimei University, Xiamen 361021, China)

Based on the preliminary analysis results of the indeterminate event stream that generated by the sensors and control purpose equipment of CPS, the adaptive dynamic Bayesian network and parallel Markov decision process model were used to support the proactive complex event processing. In order to resolve the vast state space issue of Markov decision process for large CPS, states partition and reward decomposition methods were proposed to parallel the decision making process. The experimental result based on the simulation of traffic network shows that proposed method can process event stream effectively and has favorable scalability.

CPS, big data, complex event processing,Markov decision process

s: The National Natural Science Foundation of China(No.61173036, No.61371116), Foundation of Fujian Educational Committee (No.JA14186)

TP391

A

10.11959/j.issn.1000-436x.2016183

2016-04-07；

2016-06-26

國家自然科學基金資助項目（No.61173036，No.61371116）；福建省教育廳基金資助項目（No.JA14186）

耿少峰（1981-），男，河南臺前人，湖南大學博士生，主要研究方向為 CPS、復雜事件處理等。

王永恒（1973-），男，河北霸州人，湖南大學副教授，主要研究方向為物聯網、數據挖掘等。

李仁發（1957-），男，湖南宜章人，湖南大學教授、博士生導師，主要研究方向為CPS、嵌入式系統和無線傳感網等。

張佳（1980-），女，河南新鄉人，集美大學講師，主要研究方向為物聯網、嵌入式系統等。