原作者：CoffeeOneSugar

翻譯：劉斌華

在我之前發表的文章中，我提到我最近熱衷于Complex Event PRocessing (CEP) （復雜事件處理）。簡單來說，CEP把數據流作為輸入，根據一系列預定義的規則，把數據（或部分數據）重定向給監聽者們；又或者是當發現數據中的隱含的模式（Pattern）時，觸發事件。在大量數據被產生出來并需要進行實時地分析的場景下，CEP特別有用。

有一個很不錯的軟件項目，可以讓你做到這一點，叫做ESPER。你可以在這里找到該項目的網站。Esper向程序員提供一個稱為EPL的語言，有些類似于SQL語言，它可以很好地用于對規則和模式的配置，這些規則和模式將被應用到數據流上。

Esper附帶的文檔是相當完整的，但缺乏實際的例子，這使得它看起來難以被使用。所以這里給出一個Esper的5分鐘指導。雖然下面的例子是用java編寫，但是其實Esper是同時支持Java和C#的。我這里假設你已經下載了Esper，如果沒有，可以通過點擊這個鏈接來完成。解壓剛才下載的文件后，你應該在你磁盤的某個地方可以找到一個叫esper-3.1.0的文件夾。（譯者：這篇文章比較早了，現在esper最新版本是5.2.0）

在開始前，你需要添加幾個庫文件到你的工程中，當然，esper-3.1.0.jar是其中之一，你也需要另外其他4個第三方的庫文件，他們可以在esper-3.1.0.jar/esper/lib文件夾中找到。

現在開始我們的5分鐘吧。在你把需要分析的數據丟到CEP引擎的管道中之前，需要把這些數據結構化到對象當中去。讓我們用一個簡單的例子，寫一個類（或者叫它bean）來描述給定時間的某個股票的價格：

import java.util.Date;    public static class Tick {        String symbol;        Double price;        Date timeStamp;         public Tick(String s, double p, long t) {            symbol = s;            price = p;            timeStamp = new Date(t);        }        public double getPrice() {return price;}        public String getSymbol() {return symbol;}        public Date getTimeStamp() {return timeStamp;}         @Override        public String toString() {            return "Price: " + price.toString() + " time: " + timeStamp.toString();        }    }

它有3個屬性：股票代碼，價格和時間戳。在我們開始生成數以億計的數據前，我們需要通知引擎它需要處理哪些對象，這是通過在實例化CEP引擎時，使用一個Configuration對象來實現的：

import com.espertech.esper.client.*; public class main {    public static void main(String [] args){     //The Configuration is meant only as an initialization-time object.    Configuration cepConfig = new Configuration();    // We register Ticks as objects the engine will have to handle    cepConfig.addEventType("StockTick",Tick.class.getName());    // We setup the engine    EPServiceProvider cep = EPServiceProviderManager.getProvider("myCEPEngine",cepConfig);   }}

作為測試目的，我們現在創建一個函數來生成隨機的數據，并且把它們丟到CEP引擎當中，我們把這個函數叫做“GenerateRandomTick”，它把EPRuntime對象作為參數，這個對象用于把事件傳遞給CEP引擎：

import java.util.Random;import com.espertech.esper.client.*; public class exampleMain {   private static Random generator=new Random();    public static void GenerateRandomTick(EPRuntime cepRT){    double price = (double) generator.nextInt(10);    long timeStamp = System.currentTimeMillis();    String symbol = "AAPL";    Tick tick= new Tick(symbol,price,timeStamp);    System.out.println("Sending tick:" + tick);    cepRT.sendEvent(tick);}     public static void main(String[] args) {     //The Configuration is meant only as an initialization-time object.      Configuration cepConfig = new Configuration();     cepConfig.addEventType("StockTick",Tick.class.getName());       EPServiceProvider cep=EPServiceProviderManager.getProvider("myCEPEngine",cepConfig);       EPRuntime cepRT = cep.getEPRuntime();    }}

現在，我們有了一個可以工作的CEP引擎，和不斷輸入的虛假的數據，是時候來創建我們的第一條規則了，用Esper的說法，我們的第一條EPL語句。要這么做，我們需要請引擎的管理員記錄我們的語句。然后，CEP引擎會根據EPL語句的定義，過濾它收到的數據，當數據滿足語句中的選擇條件或者模式時，觸發事件。

public static void main(String[] args) {    //The Configuration is meant only as an initialization-time object.    Configuration cepConfig = new Configuration();    cepConfig.addEventType("StockTick",Tick.class.getName());    EPServiceProvider cep = EPServiceProviderManager.getProvider("myCEPEngine",cepConfig);    EPRuntime cepRT = cep.getEPRuntime();     // We register an EPL statement    EPAdministrator cepAdm = cep.getEPAdministrator();    EPStatement cepStatement = cepAdm.createEPL("select * from " +                                "StockTick(symbol='AAPL').win:length(2) " +                                "having avg(price) > 6.0");  }

這里，我們的規則設置為：每當最近的2次數據的平均值大于6.0時，觸發事件。

下一步，主要是創建一個監聽者并把它和我們的選擇規則產生的事件關聯起來。可以這么做：

cepStatement.addListener(new CEPListener());

這里有不同的方式來實現監聽者，下面的是其中最簡單的一種。這里監聽者只是簡單地把它從引擎中收到的對象打印出來：

public static class CEPListener implements UpdateListener { public void update(EventBean[] newData, EventBean[] oldData) {         System.out.println("Event received: "                            + newData[0].getUnderlying());    }}

到目前為止，看上去還不錯?，F在是測試我們的代碼的時候了。讓我們生成一些數據，看一切能否正常工作?？梢栽趍ain函數中添加一下行來做到：

for(int i = 0; i< 5; i++)    GenerateRandomTick(cepRT);

現在所有的代碼看上去如下所示（我把Tick類和入口函數放在一起，這樣你就能把它們復制粘貼到一個文件并運行它們）

import com.espertech.esper.client.*;import java.util.Random;import java.util.Date; public class exampleMain {     public static class Tick {        String symbol;        Double price;        Date timeStamp;         public Tick(String s, double p, long t) {            symbol = s;            price = p;            timeStamp = new Date(t);        }        public double getPrice() {return price;}        public String getSymbol() {return symbol;}        public Date getTimeStamp() {return timeStamp;}         @Override        public String toString() {            return "Price: " + price.toString() + " time: " + timeStamp.toString();        }    }     private static Random generator = new Random();     public static void GenerateRandomTick(EPRuntime cepRT) {         double price = (double) generator.nextInt(10);        long timeStamp = System.currentTimeMillis();        String symbol = "AAPL";        Tick tick = new Tick(symbol, price, timeStamp);        System.out.println("Sending tick:" + tick);        cepRT.sendEvent(tick);     }     public static class CEPListener implements UpdateListener {         public void update(EventBean[] newData, EventBean[] oldData) {            System.out.println("Event received: " + newData[0].getUnderlying());        }    }     public static void main(String[] args) { //The Configuration is meant only as an initialization-time object.        Configuration cepConfig = new Configuration();        cepConfig.addEventType("StockTick", Tick.class.getName());        EPServiceProvider cep = EPServiceProviderManager.getProvider("myCEPEngine", cepConfig);        EPRuntime cepRT = cep.getEPRuntime();         EPAdministrator cepAdm = cep.getEPAdministrator();        EPStatement cepStatement = cepAdm.createEPL("select * from " +                "StockTick(symbol='AAPL').win:length(2) " +                "having avg(price) > 6.0");         cepStatement.addListener(new CEPListener());        // We generate a few ticks...        for (int i = 0; i < 5; i++) {            GenerateRandomTick(cepRT);        }    }}

Output:

log4j:WARN No appenders could be found for logger (com.espertech.esper.epl.metric.MetricReportingPath).log4j:WARN Please initialize the log4j system properly.Sending tick:Price: 6.0 time: Tue Jul 21 01:11:15 CEST 2009Sending tick:Price: 0.0 time: Tue Jul 21 01:11:15 CEST 2009Sending tick:Price: 7.0 time: Tue Jul 21 01:11:15 CEST 2009Sending tick:Price: 4.0 time: Tue Jul 21 01:11:15 CEST 2009Sending tick:Price: 9.0 time: Tue Jul 21 01:11:15 CEST 2009Event received: Price: 9.0 time: Tue Jul 21 01:11:15 CEST 2009

正如你看到的，只有最后兩行數據平均值大于6，因此只有一個事件最終被引擎觸發了。相當不錯！

Oh，你或許擔心輸出中的第一行，是的，這里還有一點小問題。事實上，Esper使用的日志生成包log4j導致了這個警告。它是可以通過一個叫log4j.xml的文件來配置的，你可以在esper-3.1.0/examples下的某個例子中的/etc目錄下找到它。我不認為給我們所有的程序都弄一個xml配置文件是個好主意，所以我們在以下代碼中，用點技巧來配置我們的logger，在你的文件開始處加入些import和代碼：

import org.apache.log4j.ConsoleAppender;import org.apache.log4j.SimpleLayout;import org.apache.log4j.Level;import org.apache.log4j.Logger;//and this in the main function before the rest of your code:public static void main(String [] args){       SimpleLayout layout = new SimpleLayout();      ConsoleAppender appender = new ConsoleAppender(new SimpleLayout());      Logger.getRootLogger().addAppender(appender);      Logger.getRootLogger().setLevel((Level) Level.WARN);(...)

5分鐘到此結束。

下一篇文章中，我將更深入一些來探索EPL語句，提供一些代碼來連接兩個引擎實現所謂的“事件精化”（event refinement）（譯者：好像之后作者再也沒有更新過了，所以，不要指望后續了：））

原文地址：https://coffeeonesugar.Wordpress.com/2009/07/21/getting-started-with-esper-in-5-minutes/。劉斌華原創翻譯，轉載請注明出處