Esper学习之十EPL语法

2014是新的一年，正好也是本人的本命年。既然是本命年，看来今年也是本 人兴旺之年了。开了个小玩笑，同时也祝各位同行今年少调bug多涨工资,这才是最实际的。年前的最后一篇说的是子查询和join ，基本上epl的大部分简单 语法都说完了。之前有朋友问我epl怎么和数据库交互，正好今天这篇就是来专 门解释这个 问题。但是要提醒各位，本篇只是说明了在 epl中如何与数据库交 互，并且只能算是简单的交互。而高级的用法会在 esperio里有详细的指导（esperio的文档可在esper的官网找到）。在esper的文档中，epl访问数据库的配置放在了比较靠后的位 置，不过为了方便各位学习，这里会先说明和数据库交互的相关配置，然后再说epl怎么访问数据库。配置文件在官方esper包的etc文件夹下，大家可以参考着学习1. 连接数据库a. JNDI获取连接配置如下：html view pla in copy -Q1. vdatabase-refereneename=mydb12. vdatasource-c onn eeti oncon text-lookup-n ame=java:comp/e nv/jdbc/mydb3. 4. 5. 6. database-referenee 的name是要连接的数据库名字，其余的配置可参考JNDI的文档使用方法：java view pla in copy 一1. if(envProperties.size()2. in itialCo ntext=3. else 4. in itialC on text=5. 6. DataSource dataSource lookupName);7. Connection connection 0) new InitialContext(envProperties);new InitialContext();=(DataSource)in itialC on text.lookup(=dataSource.getC onnection();更多内容可参考JNDI的文档b. 从连接池获取连接配置如下：(以dbcp为例)html view pla in copy1. vdatabase-refereneename=mydb32. 3.4.venv -propertyname=username value=myusername/5.venv -propertyn ame=password value=mypassword/6.venv -propertyn ame=driverClassNamevalue =com.mysql.jdbc.Driver/7.venv -propertyn ame=urlvalue =jdbc:mysql:/localhost/test/8.venv -propertyname=i ni tialSizevalue =2/9.venv -propertyn ame=validatio nQueryvalue =select1 from dual/10.v/datasourcefactory-co nn ectio n11.相同的配置可以使用esper的api达到同样的效果。代码如下:javaview pla in copy1.Propertiesprops =new Properties();2.props.put(user name.myusername);3.props.put(password,mypassword);4. props.put(driverClassName, com.mysql.jdbc.Driver);5. props.put(url, jdbc:mysql:/localhost/test);6. props.put(initialSize,2);7. props.put(validati on Query,select 1 from dual);8.8. ConfigurationDBRefconfigDB = new ConfigurationDBRef();9. /BasicDataSourceFactory is an Apache DBCP import10. c on figDB.setDataSourceFactory(props,BasicDataSourceFactory.class.getName();11. c on figDB.setC onn ectio nLifecycleE nu m(C on figurati on DBRef.C onn ecti on LifecycleE num.POOLED);12. Configurationconfiguration= new Configuration();13. c on figurati on .addDatabaseRefere nce(mydb3,con figDB);同样，也可以自己实现数据源。示例如下：javaview pla in copyr-1. configDB.setDataSourceFactory(props,MyOwnDataSourceFactory.class.getName();2.3. classMyOwnDataSourceFactory 4. public static DataSource createDataSource(Properties pr operties) 5. return new MyDataSourceImpl(properties);6. 7. c. JDBC获取连接前提是要将对应的jdbc驱动假如classpathhtml view pla in copy 81. 2. 3.vconnection-arg4.vconnection-argn ame=user value =myuser/n ame=password value =mypassword/5.vconnection-argname=somearg value =someargvalue/6.v/driverma nager-c onnection7. v/database-reference注 意：drivermanager-connection 中的 user和 password 属性必须填写，即使 增加了 connection-arg 参数也不 行。所以实际上 connection-arg 的user和 password是不需要写的。这点我觉得 esper做的不够人性化。d. 其他关于数据库连接的配置下面是一些和数据库交互的配置，更多配置可参考Javadochtml view pla in copy二1. 2. .con figuredata source or driver man ager sett in gs.3. 4. v/database-reference下面是关于连接的生命周期的配置html view pla in copy -1. vdatabase-referenee name=mydb22. .con figuredata source or driver man ager sett in gs.3. vconnection-lifecyclevalue=pooled/4. v/database-reference如果参数值为pooled，当配置了连接池，则会将每次获取的连接还给连接池。 若没配置连接池，则每次获取的连接用完后就关闭。如果参数值为retain，则会将连接缓存到esper引擎中，这个epl用完后，另 一个epl可以接着用2. 查询结果缓存策略EPL和数据库交互时会产生查询结果，所以引擎若能缓存查询结果将大大提高执 行效率，因此esper提供了两种缓存模式。a. LRU CacheLRU即least-recently-used,中文释义为“最近最少使用”，学过OS的应该知道内存缓存策略里也有这个算法，不明白的请自行搜索。配置如下：htmlview pla in copy _ ；吐1. vdatabase-refereneename=mydb2. .con figuredata source or driver man ager sett in gs.3. 4. v/database-referencesize的参数值表示最多能缓存多少条查询结果，而不是大小b. Expiry-time Cache条查询结果)，并且定该策略为每一次的查询结果设置了存活期 期清理过期的查询结果。配置如下：html view pla in copy -Q1. vdatabase-refereneename=mydb2. .con figuredata source or driverman ager sett in gs.3. vexpiry-time-cachemax-age-sec on ds=60purge-i nterval-seco nds=120 ref-type=soft/4. v/database-referencemax-age-seconds表示存活时间，purge-interval-seconds表示每隔多久清理一次过期的内容，两者单位都是秒。ref-type 有三个参数值：weak, soft ，hard。表示查询结果的引用级别，JVM 垃圾回收的时候，会根据此参数决定何时释放缓存。具体解释如下：1) .weak表示弱引用，JVM在垃圾回收的时候将清除所有的缓存，释放内存。2) .soft表示软引用，JVM在垃圾回收的时候，只有当所有的弱引用都被回收了 才会清除所有的缓存并释放空间。3) .hard表示强引用，JVM的垃圾回收不会清除缓存，所以引擎将按照规定的存 活期和清理时间管理缓存。3.Colu mn Change Case通常情况下，表字段是大小写不敏感的，但是也有设置为小写敏感的情况， 我们 可以通过设置使得查询返回的列结果为大写或者小写。配置如下：html view pla in copy4.SQL Types Mappi ng默认的数据库字段类型映射可以满足需求，不过想修改也是可以的。配置如下:html view pla in copy二1. vsql-types-mappingsql-type=2 java-type=int /sql-type表示数据库字段类型，这里的2映射了具体类型，可在java.sql.Types 类中查到，并且这个类里包含了大部分的数据库字段类型。java-type表示对应的java数据类型，大小写不敏感。以上就是EPL和数据库交互的相关配置，下面来讲解EPL是怎么和数据库交互的EPL和数据库交互有两个前提，一是 JDBC驱动能够预编译sql，而是JDBC驱动 能获取数据库的元数据。5. Joi ning SQL Query Results通常我们想要的一种交互方式是：输入某个事件到引擎，然后引擎把事件的某个 属性作为sql的查询条件交给JDBC驱动，执行sql。正好esper为此提供了相 应的解决办法，参看语法：plai n view plai ncopy 1. sql:database_name parameterized_sql_query sql是关键字不可少，parameterized_sql_query 为sql语句，只与具体的 DB 有关，无关esper，所以数据库的那些个函数都可以用。先看一个简单的例子： plai nview plai ncopy 二1. select custId,cust_name from CustomerCallEvent,sql:MyCustomerDB select cust_name from Customer where cust_id =$custld 引擎接收CustomerCallEvent事件，将事件的custId属性作为查询值，执行 MyCustomerDB据库的Customer表，其中查 询条件为Customer的cust_id 字 段值存在，然后返回相应的custId属性值和cust_name字段值给监听器 该语法有几点需要注意：a. sql需要用单引号或者双引号引起来，然后再用方括号括起来。b. $expressio n中可以是事件属性，可以是变量、常量等，也可以是用户自定 义的函数。例如：plai nview plai ncopy 1. selectfromLimitEve ntle,2.sql:MyCustomerDB selectcust_name from Customer where3. amount $max(varLowerLimit, MyLib.getLimit(le)c. join的事件可以使用view，但是sql不可使用。不过可以将sql的查询结果 通过insert into输出到另外的事件，然后再使用 view。例如：plai nview plai ncopy_1. select customerId, customerName from CustomerCallEvent.win: time(30 sec) as cce,2. sql:MyCustomerDB select cust_id as customerId, cust_ name as customerName3. from Customer4. where cust_id = $cce.custld as cqd. 可以用as为表的字段设置别名，例如：plainview plaincopy C :1. select custId, custName from CustomerCallEvent,sql:MyCustomerDB selectcust_name as custName from Customer where cust_id = $custId e. 当使用事件的属性作为查询值是，属性名不要和字段名重名，否则会报错, esper无法识别f. join的sql语句没有限制，并且可以使用 where子句。例如：plai nview plai ncopy1.2.3.4.5.select symbol, symbolDesc from OrderEventas orders,sql:My_Oracle_DB select ce as refere nee, sql:My_MySQL_DB select as historywhere refere nce.symbol and history.symbol :symbolDesc from SymbolReferen orderList from orderHistory=orders.symbolorders.symbol除了普通的join ，EPL也支持outer join sql 语句，语法也没有什么改变。例 如：plai nview plai ncopy 1. select custId, custName from2. CustomerCallEve ntas cce3. left outer joi n4. sql:MyCustomerDB select cust_id,cust_ name as custName from Customer where cust_id = $cce.custld as cq5. on cce.custId = cq.cust_id6. Usi ng Patter ns to Request Data除了通过传递外部数据查询数据库，也可以用pattern定时或者以固定频率查询 数据库。例如：plainview plaincopy CY ：1. insertinto NewOrders2. selectorderId, orderAmount from3. patter n everytimer:i nterval(5sec),4. sql:MyCustomerDB select orderId, orderAm ount from NewOrderspattern语法之后再说，这里只让大家知道有这么一个用法。7. Polling SQL Queries via APIEsper提供了 API直接执行EPL来达到访问数据库的目的。请看下面的代码：javaview pla in copy1. package example;2.2. importcom.espertech.esper.client.Configuration;3. importcom.espertech.esper.client.EPAdministrator;4. importcom.espertech.esper.client.EPRuntime;5. importcom.espertech.esper.client.EPServiceProvider;6. importcom.espertech.esper.client.EPServiceProviderManager;7. importcom.espertech.esper.client.EPStatement;8. importcom.espertech.esper.client.EventBean;10.11.import java.util.Iterator;12.13. /*14. * Created by Luo nan qin on 4/17/14.15. */16. publicclass IteratorSQLTest 17.18.publicstaticvoidmain (Stri ngargs) throws InterruptedExcepti on19.Configurationconfig = new Configuration();20.con fig.c on figure(esper.examples.cfg.xml);21.con fig.addVariable(vari,In teger.class,1);22.EPServiceProvider epService = EPServiceProviderMa nager.getDefaultProvider(co nfig);23.24.EPAdmi nistratoradmin = epService.getEPAdministrator();25.EPR un time run time = epService.getEPR un time()26.J/id=1, n ame=lu onq27.Stringepl1 = select id,name from sql:testselectid, name from test1 where id=$vari;28.29.EPStateme nt state = admi n.createEPL(epl1);30.31./Iteratoriter = state.iterator();也可以调用safeIterator 方法，该方法以线程安全方式查询DB32.while (iter.hasNext()33.Eve ntBea n eve ntBea n = iter. next();34.System.out.pri ntl n(eve ntBea n.get(id)+ +eve ntBea n.get(” name);35.36.37.执行结果：plai n view plai ncopy1. 1 luo nq8.SQL In put Parameter and Column Output Con vers ion刚才数据库配置里面有说到可以修改数据库字段类型和java数据类型的映射关系，但是那只是针对全局的设置，如果想针对EPL来设置映射关系，可以实现SQLColu mn TypeCo nversion接口，然后通过注解 Hook调用实现类。具体代码及 解释如下：java view pla in copy1. importcom.espertech.esper.clie nt.hook.SQLColu mn TypeC on text;2. importcom.espertech.esper.client.hook.SQLColumnTypeConversionJ3. importcom.espertech.esper.client.hook.SQLColumnValueContext;4. importcom.espertech.esper.clie nt.hook.SQLI nputParameterC on text;5.5. /*6. *7. * MySQLColumnTypeConverto必须为 public 类，不然无法实例化。8. * Esper 会为每一个 EPL实例，即 EPStatement 提供一个 Convertor 实例9. *10. *该例子没有做任何转换。11. * Created by Luo nan qin on 2 14.12. */13. publicclassMySQLColumnTypeConvertor implements SQLColumnTypeC onv ersi on15.16./转换列的类型17.publicClassgetColu mn Type(SQLColu mn TypeC on text context) 18.Classclazz :=con text.getColu mn ClassType();19.returnclaz z;20.21.22./转换列的值23.publicObjectgetColum nV alue(SQLColum nV alueCo ntext con text)24.Objectobj =con text.getColu mnV alue();25.returnobj;26.27.28./ 转换传入的参数值29.public Object getParameterValue(SQLInputParameterContextcontext) 30.Object objcontext.getParameterValue();31. return obj;32. 33. 34.35.package example;36.37.import com.espertech.esper.client.Configuration;38.import com.espertech.esper.client.EPAdministrator;39.import com.espertech.esper.client.EPServiceProvider;40.import com.espertech.esper.client.EPServiceProviderManager;41.import com.espertech.esper.client.EPStatement;42.import com.espertech.esper.client.EventBean;43.44.import java.util.Iterator;45.46. /*47. * MySQLColumnTypeConverto必须为 public 类，不然无法实例化。Esper会为每一个EPL提供一个Convertor实例48. *49. * Created by Luonanqin on 2/9/14.50. */51. publicclass SQLColumnTypeConversionTest 52.52. public static void main(String args) throws Int erruptedException 53. Configuration config = new Configuration();54. config.configure(esper.examples.cfg.xml);55. config.addVariable(vari,Integer.class, 1);56. EPServiceProvider epService = EPServiceProvi derManager.getDefaultProvider(config);58.57. EPAdministrator admin = epService.getEPAdmin istrator();58. /id=1, name=luonq59. String epl1 = Hook(type=HookType.SQLCOL, hook= + MySQLColumnTypeConvertor.class.getName()62.sql:testselect+ )selectid,name fromid,name from test1 where id=$vari;63.System.out.pri ntl n(epl1);64.EPStateme nt state1= adm in. createEPL(epl1);65.66.IteratorvEve ntBea niter = state1.iterator()J67.while (iter.hasNext()68.Eve ntBea n eve ntBea n = iter. next();69.System.out.pri ntl n(eve ntBea n.get(id)+ + eve ntBea n.get(” name);70.71.72.执行结果:plai nview plai ncopy1. Hook(type=HookType.SQLCOL, hook=example.MySQLColumnTypeConvertor)select id, name from sql:testselect id,name from test1 where id=$vari2. 1 luo nq9.SQL Row POJO Con version刚才说的列类型的转换以及列结果的转换，只是普通的转换。Esper还支持表的查询结果按行转换，比如说转换为 POJO而不像之前那样只能针对每一个字段 结果单独进行转换。用法也是通过 Hook注解来调用转换类。代码如下：javaview pla in copy 1. importjava.sql.ResultSet;2. importjava.sql.SQLException;3.3. importcom.espertech.esper.client.hook.SQLOutputRowConversion;4. importcom.espertech.esper.clie nt.hook.SQLOutputRowTypeC on text6. import t;7.8./*com.espertech.esper.client.hook.SQLOutputRowValueContex9. * Created by Luonanqin on 2/10/14.10. */11.publicclassMySQLOutputRowConvertor implements SQLOutputRowConversion 12.13./ 每行查询结果转换后的类型14.publicClass getOutputRowType(SQLOutputRowTypeContextcontext) 15.return String.class;16.17.18./ 返回转换后的内容19.publicObject getOutputRow(SQLOutputRowValueContextcontext)20.ResultSet result = context.getResultSet();21.Object obj1 = null;22.Object obj2 = null;23.try 24.obj1 = result.getObject(id);25.obj2 = result.getObject(name);26. catch (SQLException e) 27.e.printStackTrace();28.29.30.return obj1 + and + obj2;31.32.33.34.packageexample;35.36.importcom.espertech.esper.client.Configuration;37.importcom.espertech.esper.client.EPAdministrator;38.importcom.espertech.esper.client.EPServiceProvider;39.importcom.espertech.esper.client.EPServiceProviderManager;40.importcom.espertech.esper.client.EPStatement;41.importcom.espertech.esper.client.EventBean;42.43.importjava.util.Iterator;44.45. /*46. * MySQLOutputRowConvertov泌须为 public 类，不然无法实例 化。 Esper会为每一个EPL提供一个Convertor实例47. *48. * Created by Luonanqin on 2/9/14.49. */50. publicclass SQLOutputRowConversionTest 51.51. public static void main(String args) throws Int erruptedException 52. Configuration config = new Configuration();53. config.configure(esper.examples.cfg.xml);EPServiceProvi54. config.addVariable(vari,Integer.class, 1);55. EPServiceProvider epService derManager.getDefaultProvider(config);58.57.EPAdministrator admin = epService.getEPAdministrator();59./ epl 中返回的流必须用“ *”表示，不能是之前那样写成 id 或者 name60.61.hook=62.select63.64./id=1, name=luonqString epl1 = Hook(type=HookType.SQLROW,+ MySQLOutputRowConvertor.class.getName()+ )select * from sql:testid, name from test1 where id=$vari;System.out.println(epl1);EPStatement state1 = admin.createEPL(epl1);65.66.Iterator iter= state1.iterator()67.68.while (iter.hasNext() EventBean eventBean = iter.next();69. System.out.println(eventBean.getUnderly ing();70. 71. 72. 执行结果：plai nview pla in copy1. Hook(type=HookType.SQLROWhook=example.MySQLOutputRowConvertor)selectfrom sql:testselectid,name from test1where id=$vari2. 1 and luonq以上就是EPL和数据库交互的内容，针对普通的查询需求来说还是够用的，至于insert ，update，delete我没有举例子，各位可以自己试试看可不可行。