这是我独立开发的一款企业级产品的第二个版本，正如它的名称一样，作为事件驱动业务的内核已在数家企业中落地发芽。
ℹ️ 注意：该版本简化了部分功能（包括组件、流水线执行拓扑DAG等），
并且不包含管理端服务（这个不重要）。 你可以基于该项目实现一些自己的业务场景，你也可以随时提交PR继续丰富这个项目。
在海量的实时数据计算中，夹杂了大量的人工干预过程，包括对数据本身的清洗、业务规则判断、调用第三方算法等，重复的开发和半工具式作业使人崩溃，更重要的是业务用户根本不懂这些如何使用。 所以借助傻瓜式的画布拖拽来实现一个自己想要的业务场景很关键，让实时事件决策就像搭积木一样简单，这大大降低了业务用户使用数据进行业务决策的难度，同时让数据开发人员关注数据治理本身而不是ETL。
任何需要实时分析、决策的应用场景：
- 💹 实时交易分析
- 对系统发生的每一笔交易进行分析决策，例如：客户营销、交易反欺诈等。
- 📈 物联网事件监控
- 对海量物联网设备事件进行动态处置，例如：灾害预警、设备自检等。
- 🚀 事件动作：
| 动作名称 | 功能 |
|---|---|
| 接口动作 | 支持调用第三方Restful API。 |
| 条件动作 | 支持组合条件、单一条件的逻辑判断。 |
| 邮件动作 | 支持发送邮件。 |
| 模式匹配 | 复杂事件中对不同模式的规则匹配。 |
| 脚本动作 | 支持自定义脚本（Groovy）。 |
- 🚀 自定义流水线
- 🚀 复杂事件（CEP）
- 🚀 单一事件
- 🚀 迟到事件处理
| 名词 | 解释 |
|---|---|
| ⚙️ 事件 | 一个独立的原子业务，包括：事件名称、事件时间、事件源、事件内容4个关键要素。 |
| 📮 渠道 | 事件来源的定义，用于对事件进行分类分级，例如：交易渠道、设备网关等。 |
| 🔗 动作 | 一个具体的执行动作，例如：发送邮件、调用接口、执行算法等。 |
| 🔄 流水线 | 一个完整的业务处理场景，由多个动作组合的复杂流程。 |
这里用一个发送邮件的事件举例。
1. 环境准备（可选）
为方便快速部署，建议使用Docker完成以下环境准备工作。
- Clickhouse
默认支持Clickhouse Sink作为实时数据仓库，推荐使用21.0及以上版本。
docker pull clickhouse/clickhouse-server
docker run -p 8123:8123 -p 3500:3500 --name clickhouse-server -e CLICKHOUSE_DB=default -e CLICKHOUSE_USER=root -e CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1 -e CLICKHOUSE_PASSWORD=123456 -e TZ=Asia/Shanghai -d clickhouse/clickhouse-serverℹ️ 初始化数据库
执行 clickhouse.init.sql 初始化CH数据库。
- Redis
用于存储部分实时缓存数据。
docker pull redis
docker run -p 6379:6379 --name redis-server -d redis- Flink
下载 Apache Flink 1.14.5 到本地目录，这里使用简单的本地运行模式。
https://archive.apache.org/dist/flink/flink-1.14.52. 编译打包
- Maven Build
git clone https://github.com/eoctet/event-senses-bot.git
cd event-senses-bot
mvn clean package3. 流水线配置
配置一个可以发送邮件的流水线：发送邮件示例
更多流水线配置示例请参考：flow-samples/Xxx.yaml
4. 服务发布
发布Flink任务。
./bin/flink run-application event-senses-bot-22.0816.2.6.jar -yaml /YOUR_PATH/email-sample.yamlℹ️ TIPS
为了方便快速使用，上述步骤均使用默认的服务参数和本地部署模式。
本项目也支持其他的部署方案，如
Flink On YARN。
- 任务参数
事件任务参数定义了该任务的并行度、缓存服务、实时数据库信息。
app:
logger: "DEBUG" # 日志模式
lang: "zh-CN" # 事件跟踪日志语言类型 zh-CN/en
source:
kafka-bootstrap: "localhost:9093" # Kafka bootstrap
job:
parallelism:
default: 2 # Flink job 默认并行度
max: 100 # Flink job 最大并行度
checkpoint:
interval: 5000 # Flink job checkpoint 间隔时间（默认5秒）
storage: "/YOUR_PATH/log/checkpoint" # Flink job checkpoint 存储路径
filter:
enabled: true # 是否启用事件重复过滤
max-cache-expire: 300 # 最大缓存过期时间（默认300秒）
sink:
parallelism: 2 # SINK并行度
backup: "/YOUR_PATH/log/backup" # SINK输出备份目录
cache:
max-queue: 5 # 内存队列缓存刷新数量
interval-millis: 5000 # 内存队列缓存刷新间隔时间（默认5秒）
db:
redis: # Redis服务配置参数
cluster-enabled: false
cluster-nodes:
host: localhost
port: 6379
username:
password:
database: 0
pool:
max-total: 20
max-idle: 8
min-idle: 0
max-wait-millis: 3000
clickhouse: # Clickhouse服务配置参数
jdbc: "jdbc:clickhouse://localhost:8123/event_senses_bot"
username:
password:- 事件信息
接入的实时事件信息。
event:
name: "事件的名称" # 事件名称
code: "EVENT-5C0595EFEAA9" # 事件编号（EVENT-XXXXXXXX）
catalog: "TEST" # 事件分类
desc: "这是一个事件示例" # 事件描述
channel:
name: "测试事件渠道"
code: "CHANNEL-88DD8E18EAC7"
topic: "event_sense_topic"
format: "JSON"
groupId: "event_sense_group1"
encryption:
type: "TEXT"
key: ""
params:
- { param: "id", filter: true, nullable: true, datatype: "STRING", desc: "UID" }
- { param: "key1", datatype: "STRING", desc: "demo desc" }
- { param: "key2", datatype: "STRING", desc: "demo desc" }- 事件动作
一个完整的事件动作包含以下参数组成。
action:
name: "事件动作名称" # 事件动作名称
code: "ACTION-C77AC80CDBF3" # 事件动作编号（ACTION-XXXXXXXX）
type: "API" # 事件动作类型
parent: "0" # 事件动作前序节点的编号，0为根节点
config:
# 事件动作的配置参数
# 不同的事件动作配置参数模版是不一样的，具体请参考示例
# API:
# ...
# 事件动作的输出参数列表
# { param: "参数名称", datatype: "数据类型", desc: "描述信息" }
# 例如：
output:
- { param: "code", datatype: "LONG", desc: "" }
- { param: "data", datatype: "STRING", desc: "" }
🎁 流水线示例
所有的组件都继承了 AbstractAction 并且实现 ActionService 接口。
package pro.octet.senses.bot.action.*
除了默认支持的组件以外，你也可以自行开发一些新的组件，它们遵循一致的接口规范，例如：
//继承AbstractAction，实现execute方法
public class DemoAction extends AbstractAction {
public DemoAction(Action action) {
super(action);
}
@Override
public ExecuteResult execute() throws ActionExecutionException {
ExecuteResult executeResult = new ExecuteResult();
// ... 业务逻辑处理
// 注意：在准实时的计算中，请不要出现耗时操作、加锁操作，这会严重降低实时性能
return executeResult;
}
}关于组件执行的异常处理
如果是业务逻辑的异常，请自行
try-catch，如果是系统级的异常，请抛出到上层由引擎处理。
- 组件类型的定义
pro.octet.senses.bot.core.enums.ActionType- 组件异常的定义
pro.octet.senses.bot.exception.ActionExecutionException
Q：是否可以基于该项目实现一些自己的业务场景？
A：完全可以。
Q：是否涉及商业版权问题？
A：涉及，如果你需要进行商业开发，请联系我。
Q：我是否可以提交PR？
A：欢迎一切有想法的创意加入进来！
Q：为什么没有管理端服务？
A：本项目是事件引擎的主要实现，管理端服务并不重要，你可以基于自己的想法去重新设计它。
- 如果你有任何疑问，欢迎在GitHub Issue中提交。
