设计概念
quickmsg
# 引入依赖
<!--smqtt依赖 -->
<dependency>
<groupId>io.github.quickmsg</groupId>
<artifactId>smqtt-core</artifactId>
<version>${Latest version}</version>
</dependency>
1
2
3
4
5
6
2
3
4
5
6
# 技术架构
# 模块设计
| 模块 | 依赖 | 说明 |
|---|---|---|
| smqtt-rule-engine | jexl3 | 执行规则引擎核心api |
| smqtt-dsl | client id | 用于解析规则引擎 |
| smqtt-source | 各种数据源依赖配置 | 外部数据源 |
# 架构图
- 支持jexl3语法脚本
- 支持多种规则节点
- 支持动态Jexl脚本参数构建
# 规则引擎
# 规则引擎支持节点
- TOPIC 用于集群见转发
- PREDICATE 用于过滤节点
- LOG 用于打印日志
- KAFKA 用于发送到外部kafka
- ROCKET_MQ 用于发送到外部rocketmq
- HTTP 用于发送到外部http接口
- 其他暂未实现
# 规则引擎内置变量
| 参数 | 说明 | 必传 |
|---|---|---|
| timestamp | 消息时间 | 是 |
| clientIdentifier | client id | 是 |
| topic | topic | 是 |
| qos | qos服务等级 | 是 |
| retain | 是否保留消息 | 是 |
| msg | 消息body | 是 |
# Jexl3 语法详解
支持java方法调用
- 过滤qos为1并且topic为test/test
qos == 1 && topic.equals("test/test")1- body如果是json，过滤某个属性值
msg.属性 == 值1- 支持模板替换，构建自定义json
"{'topic':topic,'body':body,'qos':qos}"1jexl3官方文档
# 数据类型
SMQTTX相比较SMQTT来说，增加了多种数据类型可供选择，可以通过PREDICATE 节点进行数据控制
$.event.equals("publish") 获取推送消息
# 发布 （publish）
{
"messageId":1,
"topic":"test",
"qos":1,
"retain":false,
"body": {
},
"time":"2022 12-22 12:00:00",
"clientId":"A1212313"
}
1
2
3
4
5
6
7
8
9
10
11
12
2
3
4
5
6
7
8
9
10
11
12
# 订阅 （subscribe）
{
"messageId":1,
"subscribeTopics":[{
"topicFilter": "topic1",
"qos": "AT_LEAST_ONCE"
}],
"time":"2022 12-22 12:00:00",
"clientId":"A1212313"
}
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# 连接（connect）
{
"clientIp": "192.168.0.100",
"nodeIp": "127.0.0.1",
"version": "MQTT_3_1",
"keepalive": 120,
"cleanSession": false,
"auth": {
"username": "smqtt",
"password": "smqtt"
},
"will": {
"isRetain": false,
"willTopic": "willTest",
"willQos": 1,
"willMessage": ""
},
"time": "2022 12-22 12:00:00",
"clientId": "A1212313"
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 取消订阅 （unsubscribe）
{
"messageId":1,
"subscribeTopics":[{
"topicFilter": "topic1",
"qos": "AT_LEAST_ONCE"
}],
"time":"2022 12-22 12:00:00",
"clientId":"A1212313"
}
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# 断开连接（disconnect）
{
"time":"2022 12-22 12:00:00",
"clientId":"A1212313"
}
1
2
3
4
2
3
4
# 关闭 (close)
{
"messageId":1,
"reason":"",
"time":"2022 12-22 12:00:00",
"clientId":"A1212313"
}
1
2
3
4
5
6
2
3
4
5
6