Esper學(xué)習(xí)之十EPL語(yǔ)法(六)

2014是新的一年，正好也是本人的本命年。既然是本命年，看來(lái)今年也是本人興旺之年了。。。開(kāi)了個(gè)小玩笑，同時(shí)也祝各位同行今年少調(diào)bug多漲工資，這才是最實(shí)際的。

年前的最后一篇說(shuō)的是子查詢和join，基本上epl的大部分簡(jiǎn)單語(yǔ)法都說(shuō)完了。之前有朋友問(wèn)我epl怎么和數(shù)據(jù)庫(kù)交互，正好今天這篇就是來(lái)專門解釋這個(gè)問(wèn)題。但是要提醒各位，本篇只是說(shuō)明了在epl中如何與數(shù)據(jù)庫(kù)交互，并且只能算是簡(jiǎn)單的交互。而高級(jí)的用法會(huì)在esperio里有詳細(xì)的指導(dǎo)（esperio的文檔可在esper的官網(wǎng)找到）。

在esper的文檔中，epl訪問(wèn)數(shù)據(jù)庫(kù)的配置放在了比較靠后的位置，不過(guò)為了方便各位學(xué)習(xí)，這里會(huì)先說(shuō)明和數(shù)據(jù)庫(kù)交互的相關(guān)配置，然后再說(shuō)epl怎么訪問(wèn)數(shù)據(jù)庫(kù)。配置文件在官方esper包的etc文件夾下，大家可以參考著學(xué)習(xí)。

1.連接數(shù)據(jù)庫(kù)a.JNDI獲取連接配置如下：[html]\o"viewplain"viewplain\o"copy"copy<database-reference

name="mydb1">

<datasource-connection

context-lookup-name="java:comp/env/jdbc/mydb">

<env-property

name="java.naming.factory.initial"

value

="com.myclass.CtxFactory"/>

<env-property

name="vider.url"

value

="iiop://localhost:1050"/

>

</datasource-connection>

</database-reference>

database-reference的name是要連接的數(shù)據(jù)庫(kù)名字，其余的配置可參考JNDI的文檔使用方法：[java]\o"viewplain"viewplain\o"copy"copyif

(envProperties.size()

>

0)

{

initialContext

=

new

InitialContext(envProperties);

}

else

{

initialContext

=

new

InitialContext();

}

DataSource

dataSource

=

(DataSource)

initialContext.lookup(lookupName);

Connection

connection

=

dataSource.getConnection();

更多內(nèi)容可參考JNDI的文檔b.從連接池獲取連接配置如下：（以dbcp為例）[html]\o"viewplain"viewplain\o"copy"copy<database-reference

name="mydb3">

<!--

For

a

complete

list

of

properties

see

Apache

DBCP.

-->

<datasourcefactory-connection

class-name="mons.dbcp.BasicDataSourceFactory">

<env-property

name="username"

value

="myusername"/>

<env-property

name="password"

value

="mypassword"/>

<env-property

name="driverClassName"

value

="com.mysql.jdbc.Driver"/>

<env-property

name="url"

value

="jdbc:mysql://localhost/test"/>

<env-property

name="initialSize"

value

="2"/>

<env-property

name="validationQuery"

value

="select

1

from

dual"/>

</datasourcefactory-connection>

<connection-lifecycle

value="pooled"/>

</database-reference>相同的配置可以使用esper的api達(dá)到同樣的效果。代碼如下：[java]\o"viewplain"viewplain\o"copy"copyProperties

props

=

new

Properties();

props.put("username",

"myusername");

props.put("password",

"mypassword");

props.put("driverClassName",

"com.mysql.jdbc.Driver");

props.put("url",

"jdbc:mysql://localhost/test");

props.put("initialSize",

2);

props.put("validationQuery",

"select

1

from

dual");

ConfigurationDBRef

configDB

=

new

ConfigurationDBRef();

//

BasicDataSourceFactory

is

an

Apache

DBCP

import

configDB.setDataSourceFactory(props,

BasicDataSourceFactory.class.getName());

configDB.setConnectionLifecycleEnum(ConfigurationDBRef.ConnectionLifecycleEnum.POOLED);

Configuration

configuration

=

new

Configuration();

configuration.addDatabaseReference("mydb3",

configDB);

同樣，也可以自己實(shí)現(xiàn)數(shù)據(jù)源。示例如下：[java]\o"viewplain"viewplain\o"copy"copyconfigDB.setDataSourceFactory(props,

MyOwnDataSourceFactory.class.getName());

...

class

MyOwnDataSourceFactory

{

public

static

DataSource

createDataSource(Properties

properties)

{

return

new

MyDataSourceImpl(properties);

}

}

c.JDBC獲取連接前提是要將對(duì)應(yīng)的jdbc驅(qū)動(dòng)假如classpath[html]\o"viewplain"viewplain\o"copy"copy<database-reference

name="mydb2">

<drivermanager-connection

class-name="com.mysql.jdbc.Driver"

url="jdbc:mysql://localhost:3306/mydb2"

user="myuser"

password="mypassword">

<connection-arg

name="user"

value

="myuser"/>

<connection-arg

name="password"

value

="mypassword"/>

<connection-arg

name="somearg"

value

="someargvalue"/>

</drivermanager-connection>

</database-reference>

注意：drivermanager-connection中的user和password屬性必須填寫，即使增加了connection-arg參數(shù)也不行。所以實(shí)際上connection-arg的user和password是不需要寫的。這點(diǎn)我覺(jué)得esper做的不夠人性化。d.其他關(guān)于數(shù)據(jù)庫(kù)連接的配置下面是一些和數(shù)據(jù)庫(kù)交互的配置，更多配置可參考Javadoc[html]\o"viewplain"viewplain\o"copy"copy<database-reference

name="mydb2">

...

configure

data

source

or

driver

manager

settings...

<connection-settings

auto-commit="true"

catalog="mycatalog"

read-only="true"

transaction-isolation="1"

/>

</database-reference>

下面是關(guān)于連接的生命周期的配置[html]\o"viewplain"viewplain\o"copy"copy<database-reference

name="mydb2">

...

configure

data

source

or

driver

manager

settings...

<connection-lifecycle

value="pooled"/><!--

retain

-->

</database-reference>

如果參數(shù)值為pooled，當(dāng)配置了連接池，則會(huì)將每次獲取的連接還給連接池。若沒(méi)配置連接池，則每次獲取的連接用完后就關(guān)閉。如果參數(shù)值為retain，則會(huì)將連接緩存到esper引擎中，這個(gè)epl用完后，另一個(gè)epl可以接著用2.查詢結(jié)果緩存策略

EPL和數(shù)據(jù)庫(kù)交互時(shí)會(huì)產(chǎn)生查詢結(jié)果，所以引擎若能緩存查詢結(jié)果將大大提高執(zhí)行效率，因此esper提供了兩種緩存模式。

a.LRUCache

LRU即least-recently-used，中文釋義為“最近最少使用”，學(xué)過(guò)OS的應(yīng)該知道內(nèi)存緩存策略里也有這個(gè)算法，不明白的請(qǐng)自行搜索。配置如下：[html]\o"viewplain"viewplain\o"copy"copy<database-reference

name="mydb">

...

configure

data

source

or

driver

manager

settings...

<lru-cache

size="1000"/>

</database-reference>

size的參數(shù)值表示最多能緩存多少條查詢結(jié)果，而不是大小

b.Expiry-timeCache

該策略為每一次的查詢結(jié)果設(shè)置了存活期（注意不是每一條查詢結(jié)果），并且定期清理過(guò)期的查詢結(jié)果。配置如下：[html]\o"viewplain"viewplain\o"copy"copy<database-reference

name="mydb">

...

configure

data

source

or

driver

manager

settings...

<expiry-time-cache

max-age-seconds="60"

purge-interval-seconds="120"

ref-type="soft"/>

</database-reference>

max-age-seconds表示存活時(shí)間，purge-interval-seconds表示每隔多久清理一次過(guò)期的內(nèi)容，兩者單位都是秒。

ref-type有三個(gè)參數(shù)值：weak，soft，hard。表示查詢結(jié)果的引用級(jí)別，JVM垃圾回收的時(shí)候，會(huì)根據(jù)此參數(shù)決定何時(shí)釋放緩存。具體解釋如下：

1).weak表示弱引用，JVM在垃圾回收的時(shí)候?qū)⑶宄械木彺妫尫艃?nèi)存。

2).soft表示軟引用，JVM在垃圾回收的時(shí)候，只有當(dāng)所有的弱引用都被回收了才會(huì)清除所有的緩存并釋放空間。

3).hard表示強(qiáng)引用，JVM的垃圾回收不會(huì)清除緩存，所以引擎將按照規(guī)定的存活期和清理時(shí)間管理緩存。

3.ColumnChangeCase

通常情況下，表字段是大小寫不敏感的，但是也有設(shè)置為小寫敏感的情況，我們可以通過(guò)設(shè)置使得查詢返回的列結(jié)果為大寫或者小寫。配置如下：[html]\o"viewplain"viewplain\o"copy"copy<column-change-case

value="lowercase"/>

4.SQLTypesMapping

默認(rèn)的數(shù)據(jù)庫(kù)字段類型映射可以滿足需求，不過(guò)想修改也是可以的。配置如下：[html]\o"viewplain"viewplain\o"copy"copy<sql-types-mapping

sql-type="2"

java-type="int"

/>

sql-type表示數(shù)據(jù)庫(kù)字段類型，這里的2映射了具體類型，可在java.sql.Types類中查到，并且這個(gè)類里包含了大部分的數(shù)據(jù)庫(kù)字段類型。java-type表示對(duì)應(yīng)的java數(shù)據(jù)類型，大小寫不敏感。

以上就是EPL和數(shù)據(jù)庫(kù)交互的相關(guān)配置，下面來(lái)講解EPL是怎么和數(shù)據(jù)庫(kù)交互的。

EPL和數(shù)據(jù)庫(kù)交互有兩個(gè)前提，一是JDBC驅(qū)動(dòng)能夠預(yù)編譯sql，而是JDBC驅(qū)動(dòng)能獲取數(shù)據(jù)庫(kù)的元數(shù)據(jù)。

5.JoiningSQLQueryResults

通常我們想要的一種交互方式是：輸入某個(gè)事件到引擎，然后引擎把事件的某個(gè)屬性作為sql的查詢條件交給JDBC驅(qū)動(dòng)，執(zhí)行sql。正好esper為此提供了相應(yīng)的解決辦法，參看語(yǔ)法：[plain]\o"viewplain"viewplain\o"copy"copysql:database_name

["

parameterized_sql_query

"]

sql是關(guān)鍵字不可少，parameterized_sql_query為sql語(yǔ)句，只與具體的DB有關(guān)，無(wú)關(guān)esper，所以數(shù)據(jù)庫(kù)的那些個(gè)函數(shù)都可以用。先看一個(gè)簡(jiǎn)單的例子：[plain]\o"viewplain"viewplain\o"copy"copyselect

custId,

cust_name

from

CustomerCallEvent,

sql:MyCustomerDB

['

select

cust_name

from

Customer

where

cust_id

=

${custId}

']

引擎接收CustomerCallEvent事件，將事件的custId屬性作為查詢值，執(zhí)行MyCustomerDB數(shù)據(jù)庫(kù)的Customer表，其中查詢條件為Customer的cust_id字段值存在，然后返回相應(yīng)的custId屬性值和cust_name字段值給監(jiān)聽(tīng)器

該語(yǔ)法有幾點(diǎn)需要注意：

a.sql需要用單引號(hào)或者雙引號(hào)引起來(lái)，然后再用方括號(hào)括起來(lái)。

b.${expression}中可以是事件屬性，可以是變量、常量等，也可以是用戶自定義的函數(shù)。例如：[plain]\o"viewplain"viewplain\o"copy"copyselect

*

from

LimitEvent

le,

sql:MyCustomerDB

['

select

cust_name

from

Customer

where

amount

>

${max(varLowerLimit,

MyLib.getLimit(le))}

']

c.join的事件可以使用view，但是sql不可使用。不過(guò)可以將sql的查詢結(jié)果通過(guò)insertinto輸出到另外的事件，然后再使用view。例如：[plain]\o"viewplain"viewplain\o"copy"copyselect

customerId,

customerName

from

CustomerCallEvent.win:time(30

sec)

as

cce,

sql:MyCustomerDB

["select

cust_id

as

customerId,

cust_name

as

customerName

from

Customer

where

cust_id

=

${cce.custId}"]

as

cq

d.可以用as為表的字段設(shè)置別名，例如：[plain]\o"viewplain"viewplain\o"copy"copyselect

custId,

custName

from

CustomerCallEvent,

sql:MyCustomerDB

['

select

cust_name

as

custName

from

Customer

where

cust_id

=

${custId}

']

e.當(dāng)使用事件的屬性作為查詢值是，屬性名不要和字段名重名，否則會(huì)報(bào)錯(cuò)，esper無(wú)法識(shí)別

f.join的sql語(yǔ)句沒(méi)有限制，并且可以使用where子句。例如：[plain]\o"viewplain"viewplain\o"copy"copyselect

symbol,

symbolDesc

from

OrderEvent

as

orders,

sql:My_Oracle_DB

['select

symbolDesc

from

SymbolReference']

as

reference,

sql:My_MySQL_DB

['select

orderList

from

orderHistory']

as

history

where

reference.symbol

=

orders.symbol

and

history.symbol

=

orders.symbol

除了普通的join，EPL也支持outerjoinsql語(yǔ)句，語(yǔ)法也沒(méi)有什么改變。例如：[plain]\o"viewplain"viewplain\o"copy"copyselect

custId,

custName

from

CustomerCallEvent

as

cce

left

outer

join

sql:MyCustomerDB

["select

cust_id,

cust_name

as

custName

from

Customer

where

cust_id

=

${cce.custId}"]

as

cq

on

cce.custId

=

cq.cust_id

6.UsingPatternstoRequestData

除了通過(guò)傳遞外部數(shù)據(jù)查詢數(shù)據(jù)庫(kù)，也可以用pattern定時(shí)或者以固定頻率查詢數(shù)據(jù)庫(kù)。例如：[plain]\o"viewplain"viewplain\o"copy"copyinsert

into

NewOrders

select

orderId,

orderAmount

from

pattern

[every

timer:interval(5

sec)],

sql:MyCustomerDB

['select

orderId,

orderAmount

from

NewOrders']

pattern語(yǔ)法之后再說(shuō)，這里只讓大家知道有這么一個(gè)用法。7.PollingSQLQueriesviaAPI

Esper提供了API直接執(zhí)行EPL來(lái)達(dá)到訪問(wèn)數(shù)據(jù)庫(kù)的目的。請(qǐng)看下面的代碼：[java]\o"viewplain"viewplain\o"copy"copypackage

example;

import

com.espertech.esper.client.Configuration;

import

com.espertech.esper.client.EPAdministrator;

import

com.espertech.esper.client.EPRuntime;

import

com.espertech.esper.client.EPServiceProvider;

import

com.espertech.esper.client.EPServiceProviderManager;

import

com.espertech.esper.client.EPStatement;

import

com.espertech.esper.client.EventBean;

import

java.util.Iterator;

/**

*

Created

by

Luonanqin

on

4/17/14.

*/

public

class

IteratorSQLTest

{

public

static

void

main(String[]

args)

throws

InterruptedException

{

Configuration

config

=

new

Configuration();

config.configure("esper.examples.cfg.xml");

config.addVariable("vari",

Integer.class,

1);

EPServiceProvider

epService

=

EPServiceProviderManager.getDefaultProvider(config);

EPAdministrator

admin

=

epService.getEPAdministrator();

EPRuntime

runtime

=

epService.getEPRuntime();

//

id=1,

name="luonq"

String

epl1

=

"select

id,

name

from

sql:test['select

id,

name

from

test1

where

id=${vari}']";

EPStatement

state

=

admin.createEPL(epl1);

Iterator<EventBean>

iter

=

state.iterator();

//

也可以調(diào)用safeIterator方法，該方法以線程安全方式查詢DB

while

(iter.hasNext())

{

EventBean

eventBean

=

iter.next();

System.out.println(eventBean.get("id")

+

"

"

+

eventBean.get("name"));

}

}

}

執(zhí)行結(jié)果：[plain]\o"viewplain"viewplain\o"copy"copy1

luonq

8.SQLInputParameterandColumnOutputConversion

剛才數(shù)據(jù)庫(kù)配置里面有說(shuō)到可以修改數(shù)據(jù)庫(kù)字段類型和java數(shù)據(jù)類型的映射關(guān)系，但是那只是針對(duì)全局的設(shè)置，如果想針對(duì)EPL來(lái)設(shè)置映射關(guān)系，可以實(shí)現(xiàn)SQLColumnTypeConversion接口，然后通過(guò)注解Hook調(diào)用實(shí)現(xiàn)類。具體代碼及解釋如下：[java]\o"viewplain"viewplain\o"copy"copyimport

com.espertech.esper.client.hook.SQLColumnTypeContext;

import

com.espertech.esper.client.hook.SQLColumnTypeConversion;

import

com.espertech.esper.client.hook.SQLColumnValueContext;

import

com.espertech.esper.client.hook.SQLInputParameterContext;

/**

*

*

MySQLColumnTypeConvertor必須為public類，不然無(wú)法實(shí)例化。

*

Esper會(huì)為每一個(gè)EPL實(shí)例，即EPStatement提供一個(gè)Convertor實(shí)例

*

*

該例子沒(méi)有做任何轉(zhuǎn)換。

*

Created

by

Luonanqin

on

2/9/14.

*/

public

class

MySQLColumnTypeConvertor

implements

SQLColumnTypeConversion{

//

轉(zhuǎn)換列的類型

public

Class

getColumnType(SQLColumnTypeContext

context)

{

Class

clazz

=

context.getColumnClassType();

return

clazz;

}

//

轉(zhuǎn)換列的值

public

Object

getColumnValue(SQLColumnValueContext

context)

{

Object

obj

=

context.getColumnValue();

return

obj;

}

//

轉(zhuǎn)換傳入的參數(shù)值

public

Object

getParameterValue(SQLInputParameterContext

context)

{

Object

obj

=

context.getParameterValue();

return

obj;

}

}

package

example;

import

com.espertech.esper.client.Configuration;

import

com.espertech.esper.client.EPAdministrator;

import

com.espertech.esper.client.EPServiceProvider;

import

com.espertech.esper.client.EPServiceProviderManager;

import

com.espertech.esper.client.EPStatement;

import

com.espertech.esper.client.EventBean;

import

java.util.Iterator;

/**

*

MySQLColumnTypeConvertor必須為public類，不然無(wú)法實(shí)例化。

Esper會(huì)為每一個(gè)EPL提供一個(gè)Convertor實(shí)例

*

*

Created

by

Luonanqin

on

2/9/14.

*/

public

class

SQLColumnTypeConversionTest

{

public

static

void

main(String[]

args)

throws

InterruptedException

{

Configuration

config

=

new

Configuration();

config.configure("esper.examples.cfg.xml");

config.addVariable("vari",

Integer.class,

1);

EPServiceProvider

epService

=

EPServiceProviderManager.getDefaultProvider(config);

EPAdministrator

admin

=

epService.getEPAdministrator();

//

id=1,

name="luonq"

String

epl1

=

"@Hook(type=HookType.SQLCOL,

hook='"

+

MySQLColumnTypeConvertor.class.getName()

+

"')select

id,

name

from

sql:test['select

id,

name

from

test1

where

id=${vari}']";

System.out.println(epl1);

EPStatement

state1

=

admin.createEPL(epl1);

Iterator<EventBean>

iter

=

state1.iterator();

while

(iter.hasNext())

{

EventBean

eventBean

=

iter.next();

System.out.println(eventBean.get("id")

+

"

"

+

eventBean.get("name"));

}

}

}

執(zhí)行結(jié)果：[plain]\o"viewplain"viewplain\o"copy"copy@Hook(type=HookType.SQLCOL,

hook='example.MySQLColumnTypeConvertor')select

id,

name

from

sql:test['select

id,

name

from

test1

where

id=${vari}']

1

luonq

9.SQLRowPOJOConversion

剛才說(shuō)的列類型的轉(zhuǎn)換以及列結(jié)果的轉(zhuǎn)換，只是普通的轉(zhuǎn)換。Esper還支持表的查詢結(jié)果按行轉(zhuǎn)換，比如說(shuō)轉(zhuǎn)換為POJO，而不像之前那樣只能針對(duì)每一個(gè)字段結(jié)果單獨(dú)進(jìn)行轉(zhuǎn)換。用法也是通過(guò)Hook注解來(lái)調(diào)用轉(zhuǎn)換類。代碼如下：[java]\o"viewplain"viewplain\o"copy"copyimport

java.sql.ResultSet;

import

java.sql.SQLException;

import

com.espertech.esper.client.hook.SQLOutputRowConversion;

import

com.espertech.esper.client.hook.SQLOutputRowTypeContext;

import

com.espertech.esper.client.hook.SQLOutputRowValueContext;

/**

*

Created

by

Luonanqin

on

2/10/14.

*/

public

class

MySQLOutputRowConvertor

implements

SQLOutputRowConversion

{

//

每行查詢結(jié)果轉(zhuǎn)換后的類型

public

Class

getOutputRowType(SQLOutputRowTypeContext

context)

{

return

String.class;

}

//

返回轉(zhuǎn)換后的內(nèi)容

public

Object

getOutputRow(SQLOutputRowValueContext

context)

{

ResultSet

result

=

context.getResultSet();

Object

obj1

=

null;

Object

obj2

=

null;

try

{

obj1

=

result.getObject("id");

obj2

=

result.getObject("name");

}

catch

(SQLException

e)

{

e.printStackTrace();

}

return

obj1

+

"

and

"

+

obj2;

}

}

package

example;

import

com.espertech.esper.client.Configuration;

import

com.espertech.esper.client.EPAdministrator;

import

com.espertech.esper.client.EPServiceProvider;

import

com.espertech.esper.client.EPSer