基于时间累积 :select * from Withdrawal.win:time_batch(4 sec) 基于数量累积 :select * from Withdrawal.win:length_batch (5) Select * from Withdrawal.win:length(5)

各类大数据量、高实时性系统

金融分析 RFID 事件处理 流程监控 位置服务 欺诈检测
基于 JAVA 的 EDA

SEP

JMS ESB 无相关标准 有产品： IBM 、 Weblogic ， Esper

ESP&CEP

JMS

点对点
JMS

发布订阅
ESB （ Enterprise Service Bus ）

轮询数据库 浪费大量数据空间 实现复杂，不能简洁的实现临时逻辑和关联事件的因 果关系等。 非标准的 event-processing language 未对临时数据流的优化 不能连续处理事件流

分布式缓存或 JINI 网络空间


规则引擎 (Rule engines) ＋ JMS


什么是事件

服务正交相关 监控

EDA

基于消息传递，松耦合
EDA VS SOA

EDA ：“发布／订阅”

通过特定模式来对业务事件作出响应 通常耦合度比较低 允许传递粗力度的事件

SOA ：请求／响应

结合 SOA 和 EDA

SOA

垂直（ Vertical ）系统的请求／响应处理 横向（ Horizontal ）系统通讯





Events: – A1 B1 C1 B2 A2 D1 A3 B3 E1 A4 F1 B4 • pattern [ A -> every B ] – {A1,B1}, {A1,B2}, {A1,B3}, {A1,B4}, pattern [ every A -> every B ] – {A1, B1}, {A1, B2}, – {A1, B3}, {A2, B3}, {A3, B3}, – {A1, B4}, {A2, B4}, {A3, B4} and {A4, B4}
Esper sample

Listener and Engine
import net.esper.client.*; // Get engine instance and register statement EPServiceProvider engine = EPServiceProviderManager.getDefaultProvide r(); EPStatement statement = engine.getEPAdministrator().createEQL ("..."); // Attach a listener statement.addListener(new UpdateListener() { public void update(EventBean[] newEvents, EventBean[] oldEvents) { // Handle complex event ... }
API 概述

EPServiceProvider

引擎 线程 时间 流 Statment/Queries 事件查询语言 :EQL Listener POJI

EPStatement:


UpdateListener:

事件

事件可以是：

POJO Key-value 对 (java.util.map) XML(org.w3c.dom.Node)



事件例子
RFID 设备跟踪


位置信息

设备标识 X Y 基站信息变更 根据用户的位置的变更，定向推送所在区域的服务信 息：商场、电影院、公交站 . 统计／分析用户的日常行为规律。

Use-case


EDA 基本原理

EDA(Even-Driven Architect) 松耦合 基于事件 基于消息排队的架构 异步通讯 事件处理模型
Esper sample

ESP/CEP Statement （ EPL)
/ A statement can produce implicit events insert into CountZone select zone, count(*) as cnt from LocationReport.std:unique('assetId') where assetId in (1, 2, 3) group by zone
事件处理中间件原理和应用
Event-Driven Application Server
李志强 (li@) 湖南拓维信息系统股份有限公司 研发中心 2008/07/22
主要内容和目的
学习、研究事件流 (Event Stream) 和负责事 件处理 (ESP/CEP ）的基本概念 理解事件驱动应用服务器的角色和基本原理 基于 Esper 的应用和开发

指定数量范围内的事件

CEP 查询

定义事件匹配模式 标识复杂事件流 模式关键词




模式

周期性／重复性 :every 逻辑操作 :and 、 or 、 not 跟着发生： -> 子查询条件表达式 :timer:within

5 秒内的所有 A 或 B 事件 ( A or B ) where timer:within (5 sec) timer:interval:

EDA ：



EDA
问题场景

需要实时、连续地分析数据，并根据历史数据处 理模式来自动检测、发现问题 高实时性：低的处理、分析、响应延迟 大数据量：



巨量数据：每秒超 100000 个请求（事件） 超过通常使用的 OLTP 系统的处理能力

高事务数：


事件处理实现方案对比

数据库

引擎 (Engine) ：独立单元（时间、线程、事件 流） 基本语句 (Statements) ： Event processing Lan guage(EPL ） 事件处理器 (Listener): 简单 java interface


Epser 产生事件

事件发送
import net.esper.client.*; // Get the same engine instance EPServiceProvider engine = EPServiceProviderManager.getDefaultProvider(); EPRuntime runtimeEngine = engine.getRuntime(); ... LocationReport event = new LocationReport(assetId,x,y,zone); runtimeEngine.sendEvent(event);
处理模式

连续处理 结果集发生改变时通知 Listerners


新的事件到达 旧事件超出结果集／范围

内建数据库
EQL(EPL)

类 SQL 语法

流 streams: 表 事件 Event: 记录 事件属性 Event Attributes: 记录字段 ESP 查询 CEP 查询


简单事件处理（ SEP ）

基于单个事件 单个事件触发响应 通常采用：



JMS

Queue: 点对点 Topic ：发布／订阅 channel ， pipe ， router etc.

EAI 模式：

事件流处理 (ESP)

基于“流”的处理 单个时间不会触发“反应” 需要分析事件流

查询 Query ：

ESP 查询


单个事件： select * from Withdrawal 时间窗口（范围）内的事件 : select count(*) from Withdrawal(zone=10).win:time (30 sec) 批量处理：在通知 Listener 前累积事件，一次通知
事件 (Event) 是有意义的状态变化： a signifi cant change in state


股票价格的变化 密码变更 最后一次服务的响应时间 XML POJO Key-value 对

事件在系统中的表述

事件的基本特征

不只是“发生什么事情” 意发生事件的不可变记录 事件要素：标识、发生时间，有意义的属性 事件间可能有某种关联：时间顺序、因果关系
目录
事件和事件处理 Esper DEMO 总结 Q&A




Esper 简介

基于 JAVA 的 ESP ／ CEP 容器

轻量级、可嵌入 开源 包括 ESP 和 CEP 商业支持 被广泛集成到商业产品 : weblogic event server

项目背景

Esper 架构






EDA 需求

事件流：

高吞吐量 高可靠性 低延迟 事件关联