Java物联网开发(二) —— 开源百万级分布式 MQTT 消息服务器EMQX
开源百万级分布式 MQTT 消息服务器EMQX
- 一. 是什么
- 1. 简介
- 2. 分类
- 3. EMQ X 消息服务器功能列表
- 二. 安装
- 1. 安装方式
- rpm安装
- docker安装
- 免安装
- 2. 目录结构
- 3. 使用websocket实现消息的收发
- 三. 使用emqx基础功能
- 1. Dashboard
- 2. 认证
- 认证介绍
- 认证方式
- 匿名认证
- 密码加盐规则与哈希方法
- EMQ X 身份认证流程
- 认证结果
- 认证操作案例
- username 认证
- Client ID 认证
- http认证
- 四. 使用emqx高级功能
- 1. 客户端SDK
- sdk-paho
- MQTT.js
- 2. 日志与追踪
- 日志级别
- 运行时修改日志级别
- 日志追踪
- 日志文件和日志滚动
- 3. ACL 发布订阅权限控制
- 功能搭建
- 测试结果
- 4. WebHook插件
- 5. 管理监控API的使用
- 6. 保留消息
- 7. 共享订阅
- 带群组的共享订阅
- 不带群组的共享订阅消息
- 均衡策略与派发 Ack 配置
- 10. 代理订阅
- 静态/内置代理订阅
- 基于Webhook和API实现动态代理订阅
- 11. 主题重写
- 12. 黑名单配置
- 13. 速率限制
- 14. 消息重传
- 15. 系统调优
- 五. 规则引擎
- 介绍及原理
- SQL语句
- 规则引擎接收案例
一. 是什么
1. 简介
MQTT属于是物联网的通信协议，在MQTT协议中有两大角色：客户端（发布者/订阅者），服务端（Mqtt broker）；针对客户端和服务端需要有遵循该协议的的具体实现，EMQ/EMQ X就是MQTT Broker的一种实现。
EMQ X 是开源百万级分布式 MQTT 消息服务器（MQTT Messaging Broker），用于支持各种接入标准 MQTT协议的设备，实现从设备端到服务器端的消息传递，以及从服务器端到设备端的设备控制消息转发。从而实现物联网设备的数据采集，和对设备的操作和控制
EMQ官网传送门(本博客基于emq x 4.1产品编写)
2. 分类
EMQ X 公司主要提供三个产品，可在官网首页产品导航查看每一种产品；主要体现在支持的连接数量、产品功能和商业服务等方面的区别：
- EMQ X Broker：EMQ X 开源版，完整支持 MQTT V3.1.1/V5.0 协议规范，完整支持 TCP、TLS、 WebSocket 连接，支持百万级连接和分布式集群架构；LDAP, MySQL, Redis, MongoDB 等扩展插件集成，支持插件模式扩展服务器功能；支持跨 Linux、Windows、macOS 平台安装，支持公有云、私有云、K8S/容器部署
- EMQ X Enterprise：EMQ X 企业版，在开源版基础上，支持物联网主流协议 MQTT、MQTT-SN、
CoAP/LwM2M、HTTP、WebSocket 一站式设备接入；JT-808/GBT-32960 等行业协议支持，基于 TCP/UDP私有协议的旧网设备接入兼容，多重安全机制与认证鉴权；高并发软实时消息路由；强大灵活的内置规则引擎；企业服务与应用集成；多种数据库持久化支持；消息变换桥接转发 Kafka；管理监控中心 - EMQ X Platform：EMQ X 平台版，EMQ X Platform 是面向千万级超大型 IoT 网络和应用，全球首选电信级物联网终端接入解决方案。千万级大容量；多物联网协议；电信级高可靠；卓越 5G 网络支持；跨云跨IDC 部署；兼容历史系统；完善的咨询服务（从咨询到运维）
产品功能对比图
3. EMQ X 消息服务器功能列表
虽然EMQ X Enterprise, EMQ X Platform性能更加强大, EMQ X Enterprise 使用说明传送门
但是收费. 本着学习的原则选择的是性能稍差的EMQ X Broker
- 完整的 MQTT V3.1/V3.1.1 及 V5.0 协议规范支持
QoS0, QoS1, QoS2 消息支持
持久会话与离线消息支持
Retained 消息支持
Last Will 消息支持 - TCP/SSL 连接支持
- MQTT/WebSocket/SSL 支持
- HTTP 消息发布接口支持
- $SYS/# 系统主题支持
- 客户端在线状态查询与订阅支持
- 客户端 ID 或 IP 地址认证支持
- 用户名密码认证支持
- LDAP、Redis、MySQL、PostgreSQL、MongoDB、HTTP 认证集成
- 浏览器 Cookie 认证
- 基于客户端 ID、IP 地址、用户名的访问控制 (ACL)
- 多服务器节点集群 (Cluster)
- 支持 manual、mcast、dns、etcd、k8s 等多种集群发现方式
- 网络分区自动愈合
- 消息速率限制
- 连接速率限制
- 按分区配置节点
- 多服务器节点桥接 (Bridge)
- MQTT Broker 桥接支持
- Stomp 协议支持
- MQTT-SN 协议支持
- CoAP 协议支持
- LwM2M 协议支持
- Stomp/SockJS 支持
- 延时 Publish ($delay/topic)
- Flapping 检测
- 黑名单支持
- 共享订阅 ($share/:group/topic)
- TLS/PSK 支持
- 规则引擎
空动作 (调试)
消息重新发布
桥接数据到 MQTT Broker
检查 (调试)
发送数据到 Web 服务
二. 安装
EMQX支持的安装方式多种多样, docker安装, rpm安装, zip安装
点击跳转到安装地址
1. 安装方式
rpm安装
下面首先演示rpm安装
# rpm安装
## rpm安装指定版本的emqx
rpm -ivh emqx-centos7-v4.0.5.x86_64.rpm
## 查询emqx安装是否成功
rpm -qa | grep emqx
## 启动emqx
empx start
empx restart
## 查看emqx运行状态
emqx_ctl status
## 关闭emqx
emqx stop
## 卸载 emqx
rpm -e emqx# 访问emqx
## 访问地址
http://emqx安装ip:18083/
## 默认用户名：admin，默认密码：public
图1: 安装命令截图
图2: 登陆emqx后的Dashboard
docker安装
需要安装docker, 如果之前没有安装请跳转至 docker教程第二章
# docker安装
## emqx版本查看地址
https://hub.docker.com/r/emqx/emqx/tags?page=1&ordering=last_updated
## 拉取指定版本镜像
docker pull emqx/emqx:v4.0.5
## 将该镜像生成对应容器并运行
docker run -tid --name emqx -p 1883:1883 -p 8083:8083 -p 8081:8081 -p 8883:8883 -p 8084:8084 -p 18083:18083 emqx/emqx:v4.0.5# 访问emqx
## 访问地址
http://emqx安装ip:18083/
## 默认用户名：admin，默认密码：public
免安装
解压zip包, 进入 bin目录下直接运行相关命令即可
bin文件夹文件截图
安装命令
# bin目录下运行
[root@docker01 bin]# ./emqx start
EMQ X Broker v4.0.5 is started successfully![root@docker01 bin]# ./emqx_ctl status
Node 'emqx@127.0.0.1' is started
emqx 4.0.5 is running[root@docker01 bin]# ./emqx stop# 访问emqx
## 访问地址
http://emqx安装ip:18083/
## 默认用户名：admin，默认密码：public
2. 目录结构
不同安装方式得到的 EMQ X 其目录结构会有所不同，具体如下:
- bin 目录
emqx、emqx.cmd：EMQ X 的可执行文件
emqx_ctl、emqx_ctl.cmd：EMQ X 管理命令的可执行文件 - etc 目录
EMQ X 通过 etc 目录下配置文件进行设置，主要配置文件包括
- data 目录
EMQ X 将运行数据存储在 data 目录下，主要的文件包括:
configs/app.*.config EMQ X 读取 etc/emqx.conf 和 etc/plugins/*.conf 中的配置后，转换为 Erlang 原生配置文件格式，并在运行时读取其中的配置。
loaded_plugins : 记录了 EMQ X 默认启动的插件列表，可以修改此文件以增删默认启动的插件。loaded_plugins 中启动项格式为{<Plugin Name>, <Enabled>}.，<Enabled>字段为布尔类型，EMQ X 会在启动时根据 的值判断是否需要启动该插件。
Mnesia 数据库是 Erlang 内置的一个分布式 DBMS，可以直接存储 Erlang 的各种数据结构
EMQ X 使用 Mnesia 数据库存储自身运行数据，例如告警记录、规则引擎已创建的资源和规则、Dashbaord用户信息等数据，这些数据都将被存储在 mnesia 目录下，因此一旦删除该目录，将导致 EMQ X 丢失所有业务数据。可以通过emqx_ctl mnesia命令查询 EMQ X 中 Mnesia 数据库的系统信息。 - log 目录
emqx.log.：EMQ X 运行时产生的日志文件
crash.dump：EMQ X 的崩溃转储文件，可以通过 etc/emqx.conf 修改配置。
erlang.log.：以 emqx start 方式后台启动 EMQ X 时，控制台日志的副本文件。
3. 使用websocket实现消息的收发
使用基于emqx的图形化ui中的websocket实现图片的收发
请求地址: http://emqx服务器ip:18083/#/websocket
创建连接
订阅以 testtopic/ 开头的消息
向主题名为 testtopic/1 的主题发送消息
这样订阅了该主题客户端就能够收到相关消息
三. 使用emqx基础功能
1. Dashboard
- 功能展示截图
- 功能描述
这些功能将在后续的使用中介绍到
- 需要注意的是可以在admin->settings->Language 中设置当前语言为中文
2. 认证
认证介绍
身份认证是大多数应用的重要组成部分，MQTT 协议支持用户名密码认证，启用身份认证能有效阻止非法客户端的连接。
EMQ X 中的认证指的是当一个客户端连接到 EMQ X 的时候，通过服务器端的配置来控制客户端连接服务器的权限。
EMQ X 的认证支持包括两个层面：
MQTT 协议本身在 CONNECT 报文中指定用户名和密码，EMQ X 以插件形式支持基于 Username、ClientID、HTTP、JWT、LDAP 及各类数据库如 MongoDB、MySQL、PostgreSQL、Redis 等多种形式的认证。
在传输层上，TLS 可以保证使用客户端证书的客户端到服务器的身份验证，并确保服务器向客户端验证服务器证书。也支持基于 PSK 的 TLS/DTLS 认证。
认证方式
EMQ X 支持使用内置数据源（文件、内置数据库）、JWT、外部主流数据库和自定义 HTTP API 作为身份认证数据源。
连接数据源、进行认证逻辑通过插件实现的，每个插件对应一种认证方式，使用前需要启用相应的插件。
客户端连接时插件通过检查其 username/clientid 和 password 是否与指定数据源的信息一致来实现对客户端的身份认证。
EMQ X 支持的认证方式：
内置数据源
Mnesia (用户名/Client ID）认证
使用配置文件与 EMQ X 内置数据库提供认证数据源，通过 HTTP API 进行管理，足够简单轻量。外部数据库
可以存储大量数据，同时方便与外部设备管理系统集成。
LDAP 认证
MySQL 认证
PostgreSQL 认证
Redis 认证
MongoDB 认证其他
HTTP 认证
JWT 认证
注意
- 更改插件配置后需要重启插件才能生效，部分认证鉴权插件包含 ACL 功能。
- 各种认证方式配置详情请点击 这里 查看
匿名认证
EMQ X 默认配置中启用了匿名认证，任何客户端都能接入 EMQ X。没有启用认证插件或认证插件没有显式允许/拒绝（ignore）连接请求时，EMQ X 将根据匿名认证启用情况决定是否允许客户端连接。
生产环境中请禁用匿名认证。
# 进入 etc/emqx.conf# 配置是否开启匿名认证(默认为true)
## Value: true | false
allow_anonymous = false配置之后, 可以发现再次使用websocket 直接连接就会失败, 必须要输入指定的用户名和密码
密码加盐规则与哈希方法
EMQ X 多数认证插件中可以启用哈希方法，数据源中仅保存密码密文，保证数据安全。
启用哈希方法时，用户可以为每个客户端都指定一个 salt（盐）并配置加盐规则，数据库中存储的密码是按照加盐规则与哈希方法处理后的密文。
以 MySQL 认证为例：加盐规则与哈希方法配置
# etc/plugins/emqx_auth_mysql.conf## 不加盐，仅做哈希处理
auth.mysql.password_hash = sha256## salt 前缀：使用 sha256 加密 salt + 密码 拼接的字符串
auth.mysql.password_hash = salt,sha256## salt 后缀：使用 sha256 加密 密码 + salt 拼接的字符串
auth.mysql.password_hash = sha256,salt## pbkdf2 with macfun iterations dklen
## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512
## auth.mysql.password_hash = pbkdf2,sha256,1000,20
EMQ X 身份认证流程
- 根据配置的认证 SQL 结合客户端传入的信息，查询出密码（密文）和 salt（盐）等认证数据，没有查询结果时，认证将终止并返回 ignore 结果
- 根据配置的加盐规则与哈希方法计算得到密文，没有启用哈希方法则跳过此步
- 将数据库中存储的密文与当前客户端计算的到的密文进行比对，比对成功则认证通过，否则认证失败
PostgreSQL 认证功能逻辑图：
认证结果
任何一种认证方式最终都会返回一个结果：
- 认证成功：经过比对客户端认证成功
- 认证失败：经过比对客户端认证失败，数据源中密码与当前密码不一致
- 忽略认证（ignore）：当前认证方式中未查找到认证数据，无法显式判断结果是成功还是失败，交由认证链下一认证方式或匿名认证来判断
认证操作案例
username 认证
Username 认证使用配置文件预设客户端用户名与密码，支持通过 HTTP API 管理认证数据。
Username 认证不依赖外部数据源，使用上足够简单轻量。使用这种认证方式前需要开启插件，我们可以在Dashboard里找到这个插件并开启。
启动
emqx_auth_username插件
配置username认证加密方式
可在etc/plugins/emqx_auth_username.conf 中更改(默认内容如下图)：# etc/plugins/emqx_auth_username.conf ## Value: plain | md5 | sha | sha256 auth.user.password_hash = sha256配置哈希方法后，新增的预设认证数据与通过 HTTP API 添加的认证数据将以哈希密文存储在 EMQ X 内置数据库中。
可以通过配置文件预设认证数据，编辑配置文件： etc/plugins/emqx_auth_username.conf
预设认证数据格式兼容 emqx_auth_clientid 与 emqx_auth_username 插件的配置格式
但是这种方式预先配置的用户信息, 是明文的没有加密, 安全性得不到保证(因此无更改可跳过2,3步)##-------------------------------------------------------------------- ## Username Authentication Plugin ##--------------------------------------------------------------------## Examples: ##auth.user.1.username = admin ##auth.user.1.password = public ##auth.user.2.username = feng@emqtt.io ##auth.user.2.password = public ##auth.user.3.username = name~!@#$%^&*()_+ ##auth.user.3.password = pwsswd~!@#$%^&*()_+## Password hash. ## ## Value: plain | md5 | sha | sha256 auth.user.password_hash = sha256安装 vscode -> 安装 rest client 插件 -> 创建新文件->选择保存(选择文件类型为 .http)
输入基本参数以及请求格式后, 点击Send Request , 右侧就会展示请求的结果
#这里使用的userName和password 是我们登陆ui时默认的用户, 也可以自己在ui界面中新建用户 @hostname = emqx所在服务器ip @port=8081 @contentType=application/json @userName=admin @password=public #############查看已有用户认证数据############## GET http://{{hostname}}:{{port}}/api/v4/auth_username HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}}########添加用户认证数据############## POST http://{{hostname}}:{{port}}/api/v4/auth_username HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}} { "username": "user", "password": "123456" }###########更改指定用户名的密码############# PUT http://{{hostname}}:{{port}}/api/v4/auth_username/user HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}} { "password": "user" }###########查看指定用户名信息############# GET http://{{hostname}}:{{port}}/api/v4/auth_username/user HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}}###########删除指定的用户信息############# DELETE http://{{hostname}}:{{port}}/api/v4/auth_username/user HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}}MQTTX客户端验证
点击跳转至下载地址
a. 新建两个链接, 一个用于发布主题消息, 一个用于订阅主题并接收消息
b.在测试主题订阅的连接上面创建消息订阅. 订阅以 testtopic/ 开头的主题
testtopic/#中, '#'为通配符, 用于匹配 以 testtopic/ 开头的主题
c.向 testtopic/1 主题发送消息
d.下图可以看到 测试消息订阅的连接 收到了 测试username认证的连接 的消息
Client ID 认证
配置和测试流程同username认证
启动client id认证插件
emqx_auth_clientid
配置client id 认证加密方式
可在etc/plugins/emqx_auth_clientid.conf 中更改(默认内容如下图)：
如无改动请跳过##-------------------------------------------------------------------- ## ClientId Authentication Plugin ##--------------------------------------------------------------------## Examples ##auth.client.1.clientid = id ##auth.client.1.password = passwd ##auth.client.2.clientid = dev:devid ##auth.client.2.password = passwd2 ##auth.client.3.clientid = app:appid ##auth.client.3.password = passwd3 ##auth.client.4.clientid = client~!@#$%^&*()_+ ##auth.client.4.password = passwd~!@#$%^&*()_+## Password hash. ## ## Value: plain | md5 | sha | sha256 auth.client.password_hash = sha256在vscode通过rest client 插件发送http请求
主要需要执行添加clientId和密码操作#这里使用的userName和password 是我们登陆ui时默认的用户, 也可以自己在ui界面中新建用户 @hostname = emqx所在服务器ip @port=8081 @contentType=application/json @userName=admin @password=public####添加clientId和密码##### POST http://{{hostname}}:{{port}}/api/v4/auth_clientid HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}} { "clientid": "test-clientid", "password": "123456" }#############获取所有详细信息######## GET http://{{hostname}}:{{port}}/api/v4/auth_clientid HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}}#############更改指定 Client ID 的密码######## PUT http://{{hostname}}:{{port}}/api/v4/auth_clientid/test-clientid HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}} { "password": "99999" }#############删除指定的client信息######## DELETE http://{{hostname}}:{{port}}/api/v4/auth_clientid/test-clientid HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}}进入MQTTX, 新建测试用连接
订阅主题, 模拟消息发送, 测试接收的消息
http认证
开启http认证, 通过客户端来进行认证, emqx通过客户端返回的响应码来判断请求成功或失败
响应码可见 认证结果 这一部分介绍
开启http认证插件
关闭上面开启的username, clientid认证插件, 这些插件默认关闭
配置http认证参数
配置文件在 /etc/plugins/emqx_auth_http.conf
注意: auth.http.auth_req 默认是本机. 如果这里不进行修改可以将下面编写好的客户端代码上传到服务器运行## Value: URL auth.http.auth_req = http://客户端认证服务器ip:8991/mqtt/auth ## Value: post | get | put auth.http.auth_req.method = post ## Value: Params auth.http.auth_req.params = clientid=%c,username=%u,password=%P
- HTTP 请求方法为 GET 时，请求参数将以 URL 查询字符串的形式传递；POST、PUT 请求则将请求参数以普通表单形式提交（content-type 为 x-www-form-urlencoded）。
- 你可以在认证请求中使用以下占位符，请求时 EMQ X 将自动填充为客户端信息：
%u：用户名
%c：Client ID
%a：客户端 IP 地址
%r：客户端接入协议
%P：明文密码
%p：客户端端口
%C：TLS 证书公用名（证书的域名或子域名），仅当 TLS 连接时有效
%d：TLS 证书 subject，仅当 TLS 连接时有效 - 推荐使用 POST 与 PUT 方法，使用 GET 方法时明文密码可能会随 URL 被记录到传输过程中的服务器日志中。
编写客户端代码
创建一个spring boot项目, 添加spring web模块即可. devtools 为热部署插件
a.pom文件代码<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build>b. 编写application.yml配置文件
需和第2步所配置的 ip 和端口一致server:port: 8991spring:application:name: emqx-democ. 编辑controller层接口
注意: 这里只验证username和passwd 没有去验证clientid, 具体逻辑可以根据自己需要来设计@RestController @RequestMapping("/mqtt") public class AuthController {private Logger log = LoggerFactory.getLogger(AuthController.class);private Map<String, String> users;//注解的作用是让项目在初始化时能够加载这个map@PostConstructpublic void init() {users = new HashMap<>();users.put("user", "123456");//实际的密码应该是密文,mqtt的http认证组件传输过来的密码是明 文，我们需要自己进行加密验证users.put("emq-client2", "123456");users.put("emq-client3", "123456");}@PostMapping("/auth")public ResponseEntity<?> auth(@RequestParam("clientid") String clientid, @RequestParam("username") String username,@RequestParam("password") String password) {log.info("emqx认证组件调用自定义的认证服务开始认证,clientid={},username={},password= {}", clientid, username, password);//为了演示方便做一个固定操作, 验证username和passwd 没有去验证clientidString value = users.get(username);if (StringUtils.isEmpty(value)) {return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);}if (!value.equals(password)) {return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);}return new ResponseEntity<Object>(HttpStatus.OK);} }d.启动类自动生成, 启动即可
注意: 如果步骤2使用的是默认, 则设置默认的地址(客户端验证代码)为emqx所在服务器ip, 需要将客户端jar上传至该服务器并运行打开mqttx , 创建http认证测试连接
参数如下, 然后点击connect
第4步连接成功后, 可以到客户端项目(进行逻辑验证)返回了 相关信息
说明通过http认证成功
四. 使用emqx高级功能
1. 客户端SDK
在实际项目中我们要针对接MQTT消息代理服务端，从而向其发布消息、订阅消息等来完成我们自己的业务逻辑的开发。EMQX针对不同的客户端语言都提供了不同的SDK工具包，可以在官网上查看并下下载
sdk-paho
基于Java语言开发的sdk工具, 通过该sdk来实现对emqx的操作
官方文档及源码地址 https://github.com/eclipse/paho.mqtt.java
这里分享下其他值得借鉴的mqtt_java_sdk : sdk_1 sdk_2 SpringBoot整合MQTT并实现异步线程调用
实现步骤
这里需要启动emqx的broker服务器, 然后访问
http://emqx所在服务器ip:18083/#/plugins查看认证插件启动情况,# 注意: 默认是开启emqx_auth_username, 但是如果配置了其他插件需要将其他认证模块插件关闭 # 关闭 emqx_auth_http, emqx_auth_clientid 插件 # 开启 emqx_auth_username 插件在基于mqtt认证 所创建的项目的基础上, 添加相关操作
# 相关操作流程大致如下 ->添加pom文件 ->yml文件添加emqx相关配置参数 ->创建读取yml文件的配置类 ->创建emqx工具类 ->创建消息发布订阅类型枚举类 ->创建回调方法 ->启动类调用@ 下面代码已发布到码云,地址如下 https://gitee.com/TimePause/emqxdemo添加pom文件
<!--mqtt客户端--><dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.2</version></dependency><!--生成配置元信息，不引入yml文件就失效了, 解决读取配置文件内容问题--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency>yml文件添加emqx相关配置参数
server:port: 8992spring:application:name: emqx-demo# 新增如下配置 mqtt:broker-url: tcp://emqx所在ip:1883client-id: demo-clientusername: userpassword: 123456创建读取yml文件的配置类
import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component;/*** MqttProperties* 创建属性配置类来加载该配置数据* @author timepause*/ @Component @ConfigurationProperties(prefix = "mqtt") public class MqttProperties {//消息服务端url//@Value("${mqtt.broker-url}")private String brokerUrl;//客户端id//@Value("${mqtt.client-id}")private String clientId;//@Value("${mqtt.username}")private String username;//@Value("{mqtt.password}")private String password;public String getBrokerUrl() {return brokerUrl;}public void setBrokerUrl(String brokerUrl) {this.brokerUrl = brokerUrl;}public String getClientId() {return clientId;}public void setClientId(String clientId) {this.clientId = clientId;}public String getUsername() {return username;}public void setUsername(String username) {this.username = username;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;}@Overridepublic String toString() {return "MqttProperties{" +"brokerUrl='" + brokerUrl + '\'' +", clientId='" + clientId + '\'' +", username='" + username + '\'' +", password='" + password + '\'' +'}';} }创建emqx工具类
import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;import javax.annotation.PostConstruct; import javax.annotation.PreDestroy;/*** 客户端封装类的代码** @author timepause*/ @Component public class EmqClient {//日志private static final Logger log = LoggerFactory.getLogger(EmqClient.class);private IMqttClient mqttClient;//回调方法才是对mqtt进行业务处理的地方@Autowiredprivate MqttCallback mqttCallback;//获取配置文件属性@Autowiredprivate MqttProperties mqttProperties;//在客户端应用程序启动后就将该方法初始化好@PostConstructprivate void init(){//MqttClientPersistence是接口 实现类有：MqttDefaultFilePersistence；MemoryPersistenceMqttClientPersistence mqttClientPersistence=new MemoryPersistence();try {mqttClient = new MqttClient(mqttProperties.getBrokerUrl(),mqttProperties.getClientId(),mqttClientPersistence);} catch (MqttException e) {log.error("初始化mqtt失败, brokerurl={},clientId={}",mqttProperties.getBrokerUrl(),mqttProperties.getClientId());}}/*** 连接服务端* @param username* @param password*/public void connect(String username, String password){MqttConnectOptions options = new MqttConnectOptions();options.setAutomaticReconnect(true);options.setConnectionTimeout(10);options.setUserName(username);options.setPassword(password.toCharArray());mqttClient.setCallback(mqttCallback);try {mqttClient.connect(options);} catch (MqttException e) {log.error("mqtt客户端连接服务器失败, 失败原因{}",e.getMessage());}}/*** 断开连接* @PreDestroy: 是对该方法的优化, 作用是在客户端连接断开时自动调用该方法*/@PreDestroypublic void disconnect(){try {mqttClient.disconnect();} catch (MqttException e) {log.error("断开连接失败, 失败原因{}",e.getMessage());}}/*** 重连方法*/public void reconnect(){try {mqttClient.reconnect();} catch (MqttException e) {log.error("重连失败, 失败原因", e.getMessage());}}/*** 消息发布* @param topic 主题名称* @param msg 消息内容* @param qos 消息发布类型* @param retain 是否保留消息*/public void publish(String topic, String msg, QosEnum qos, boolean retain){MqttMessage mqttMessage = new MqttMessage();mqttMessage.setPayload(msg.getBytes());mqttMessage.setQos(qos.value());mqttMessage.setRetained(retain);try {mqttClient.publish(topic,mqttMessage);} catch (MqttException e) {log.error("消息发布失败, errorMsg={},topic={},payload={},Qos={},retain={}",e.getMessage(),topic,msg,qos.value(),retain);}}/*** 消息发布* @param topicFilter 订阅主题* @param qos 消息发布类型*/public void subscribe(String topicFilter, QosEnum qos){try {mqttClient.subscribe(topicFilter,qos.value());} catch (Exception e) {log.error("消息订阅失败, errorMsg={},topic={},qos={}",e.getMessage(),topicFilter,qos.value());}}/*** 取消订阅* @param topicFilter 取消订阅主题*/public void unsubscribe(String topicFilter){try {mqttClient.unsubscribe(topicFilter);} catch (MqttException e) {log.error("取消订阅失败, errorMsg={},topic={}",e.getMessage(),topicFilter);}}创建消息发布订阅类型枚举类
/*** 用于枚举消息的发布类型* Qos0 : 最多一次* Qos1 : 至少一次* Qos2 : 只有一次*/ public enum QosEnum {Qos0(0),Qos1(1),Qos2(2);private final int value;//枚举类构造方法QosEnum(int value) {this.value = value;}public int value(){return this.value;}}创建回调方法
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component;/*** 对mqtt消息进行处理** @author timepause*/ @Component public class MessagCallback implements MqttCallback {private static final Logger log = LoggerFactory.getLogger(MqttCallback.class);/*** 丢失对服务端的连接后触发该方法回调，此处可以做一些特殊处理，比如重连** @param cause*/@Overridepublic void connectionLost(Throwable cause) {log.info("与broker连接丢失");}/*** 订阅到消息后的回调* 该方法由mqtt客户端同步调用,在此方法未正确返回之前，不会发送ack确认消息到broker* 一旦该方法向外抛出了异常客户端将异常关闭，当再次连接时；所有QoS1,QoS2且客户端未进行ack确认的消息都将由broker服务器再次发送到客户端** @param topic* @param message* @throws Exception*/@Overridepublic void messageArrived(String topic, MqttMessage message) throws Exception {log.info("订阅到消息;topic={},messageid={},qos={},msg={}",topic, message.getId(), message.getQos(), new String(message.getPayload()));}/** 消息发布完成且收到ack确认后的回调* QoS0：消息被网络发出后触发一次* QoS1：当收到broker的PUBACK消息后触发* QoS2：当收到broer的PUBCOMP消息后触发* @param token*/@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {int messageId = token.getMessageId();String[] topics = token.getTopics();log.info("消息发送完成,messageId={},topics={}",messageId,topics);} }启动类调用
@Autowiredprivate EmqClient emqClient;@Autowiredprivate MqttProperties mqttProperties;/*** 真正测试业务方法* 1. 连接服务端* 2. 订阅主题(通配符订阅)* 3. 开启一个线程, 每隔5秒向 testtopic/1 发送消息*/@PostConstructpublic void inited(){emqClient.connect(mqttProperties.getUsername(),mqttProperties.getPassword());emqClient.subscribe("testtopic/#", QosEnum.Qos2);new Thread(()->{while (true){emqClient.publish("testtopic/1", "publish msg"+LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME),QosEnum.Qos2,false);System.out.println("=============================");try {TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {e.printStackTrace();}}}).start();}启动测试
MQTT.js
MQTT.js是MQTT协议的客户端库，用JavaScript编写，适用于node.js和浏览器。
GitHub项目地址：https://github.com/mqttjs/MQTT.js
下面将演示基于sdk-matt.js实现消息的收发
- 在sdk-paho创建的springboot项目的resources/static 目录下, 创建html文件, 类型为html5, 名称为 index
- 编写 index.html文件代码
作用是: 创建三个div, 第一个div监听客户端收到的数据包. 第二个div显示每隔2s客户端发送的数据, 第三个div显示消息到达客户端的消息
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>mqtt.js测试</title><script src="https://cdn.staticfile.org/jquery/1.10.2/jquery.min.js" ></script><script src="https://unpkg.com/mqtt/dist/mqtt.min.js" ></script><style>div{width: 300px;height: 300px;float: left;border: 1px solid red;}</style><script type="text/javascript">$(function () {//定义连接选项对象const options = {clean: true, // 不保留回话connectTimeout: 4000, // 超时时间// 认证信息clientId: 'emqx_h5_client',username: 'user',password: '123456',}// 连接字符串, 通过协议指定使用的连接方式// ws 未加密 WebSocket 连接 8083端口// wss 加密 WebSocket 连接 8084端口// mqtt 未加密 TCP 连接// mqtts 加密 TCP 连接// wxs 微信小程序连接// alis 支付宝小程序连接const connectUrl = "ws://emqx所在服务器ip:8083/mqtt";const client = mqtt.connect(connectUrl,options);/*** mqtt.Client相关事件*///当重新连接启动触发回调client.on('reconnect', () => {$("#div1").text("正在重连.....");});//连接断开后触发的回调client.on("close",function () {$("#div1").text("客户端已断开连接.....");});//从broker接收到断开连接的数据包后发出。MQTT 5.0特性client.on("disconnect",function (packet) {$("#div1").text("从broker接收到断开连接的数据包....."+packet);});//客户端脱机下线触发回调client.on("offline",function () {$("#div1").text("客户端脱机下线.....");});//当客户端无法连接或出现错误时触发回调client.on("error",(error) =>{$("#div1").text("客户端出现错误....."+error);});//以下两个事件监听粒度细//当客户端发送任何数据包时发出。这包括published()包以及MQTT用于管理订阅和连接的包client.on("packetsend",(packet)=>{$("#div1").text("客户端已发出数据包....."+packet);});//当客户端接收到任何数据包时发出。这包括来自订阅主题的信息包以及MQTT用于管理订阅和连接的信息包client.on("packetreceive",(packet)=>{$("#div1").text("客户端已收到数据包....."+packet);});//注册监听connect事件client.on("connect",function (connack) {//成功连接上服务端之后$("#div1").text("成功连接上服务器"+new Date());//订阅 testtopic/#client.subscribe("testtopic/#",{qos:2});//每隔2秒钟发布一次setInterval(publish,2000);});function publish() {//发布数据/*** client.publish(topic,message,[options], [callback])* message: Buffer or String* options:{* qos:0, //默认0* retain:false, //默认false* dup:false, //默认false* properties:{}* }* callback:function (err){}*/const message = "h5 message "+Math.random()+new Date();client.publish("testtopic/123",message,{qos:2});$("#div2").text("客户端发布了数据:"+message);}//注册消息到达的事件client.on("message",(topic, message, packet)=>{$("#div3").text("客户端收到订阅消息,topic="+topic+";消息数据:"+message+";数据包:"+packet);});//页面离开自动断开连接$(window).bind("beforeunload",()=>{$("#div1").text("客户端窗口关闭,断开连接");client.end();})});</script>
</head>
<body><div id="div1"></div>
<div id="div2"></div>
<div id="div3"></div>
</body>
</html>
结果展示
#测试：启动项目前将启动类EmqxdemoApplication中init方法上的注解注释掉，启动后访问如下地址查看网页端的输出 代码地址: https://gitee.com/TimePause/emqxdemo 访问: http://localhost:8992/index.html可以在DashBoard上面看到下面websocket连接
可以看到这里相当于模拟第二章第3节的案例
2. 日志与追踪
日志级别
- EMQ X 的日志分 8 个等级， 由低到高分别为：
debug < info < notice < warning < error < critical < alert < emergency - EMQ X 的默认日志级别为 warning，可在 /etc/emqx.conf中修改：
log.level = warning
此配置将所有 log handler 的配置设置为 warning。
[Primary Level] -- global log level and filters/ \ [Handler 1] [Handler 2] -- log levels and filters at each handler
EMQ X 使用了分层的日志系统，在日志级别上，包括全局日志级别 (primary log level)、以及各 log handler 是负责日志处理和输出的工作进程，它由 log handler id 唯一标识，并负有如下任务：
- 接收什么级别的日志
- 如何过滤日志消息
- 将日志输出到什么地方的日志级别。
- 查看 emqx 默认安装的 log handlers:
[root@docker01 bin]# ./emqx_ctl log handlers list LogHandler(id=ssl_handler, level=debug, destination=console) LogHandler(id=file, level=warning, destination=log/emqx.log) LogHandler(id=default, level=warning, destination=console)
- file: 负责输出到日志文件的 log handler。
没有设置特殊过滤条件，所有日志消息只要级别满足要求就输出。输出目的地为日志文件。 - default: 负责输出到控制台的 log handler。
没有设置特殊过滤条件，所有日志消息只要级别满足要求就输出。输出目的地为控制台。 - ssl_handler: ssl 的 log handler。
它的过滤条件设置为当日志是来自 ssl 模块时输出。输出目的地为控制台。
总结:
- 在日志级别小节中提到的 log.level 是修改了全局的日志级别。这包括 primary log level 和各个 handlers的日志级别，都设置为了同一个值。
- Primary Log Level 相当于一个自来水管道系统的总开关，一旦关闭则各个分支管道都不再有水流通过。这个机制保证了日志系统的高性能运作。
运行时修改日志级别
可以使用 EMQ X 的命令行工具 emqx_ctl 在运行时修改 emqx 的日志级别：
修改全局日志级别：
将 primary log level 以及所有 log handlers 的级别设置为 debug：[root@docker01 bin]# ./emqx_ctl log set-level debug debug修改主日志级别：
例如，将 primary log level 设置为 debug:[root@docker01 bin]# ./emqx_ctl log primary-level debug debug修改某个log handler的日志级别：
例如，将 log handler file 设置为 debug:[root@docker01 bin]# ./emqx_ctl log handlers set-level file debug debug
日志追踪
EMQ X 支持针对 ClientID 或 Topic 过滤日志并输出到文件。在使用日志追踪功能之前，必须将 primary loglevel 设置为 debug：
./emqx_ctl log primary-level debug开启 ClientID 日志追踪，将所有 ClientID 为 emq-demo 的日志都输出到 log/my_client.log:
[root@docker01 bin]# ./emqx_ctl log primary-level debug debug[root@docker01 bin]# ./emqx_ctl trace start client emq-demo log/emq-demo.log trace clientid emq-demo successfully开启 Topic 日志追踪，将主题能匹配到 ‘testtopic/#’ 的消息发布日志输出到 log/topic_testtopic.log:
[root@docker01 bin]# ./emqx_ctl log primary-level debug debug[root@docker01 bin]# ./emqx_ctl trace start topic 'testtopic/#' log/topic_testtopic.log trace topic testtopic/# successfully查看生成的日志
日志文件和日志滚动
EMQ X 的默认日志文件目录在软件根目录下的 log目录下。
可在软件根目录下 /etc/emqx.conf 中配置：log.dir = log在文件日志启用的情况下 (log.to = file 或 both)，日志目录下会有如下几种文件:
emqx.log.N:: 包含了 EMQ X 的所有日志消息。比如emqx.log.1 , emqx.log.2 …
emqx.log.siz 和 emqx.log.idx:记录日志滚动信息的系统文件。
run_erl.log:以 emqx start 方式后台启动 EMQ X 时，用于记录启动信息的系统文件。
erlang.log.N:以 emqx start 方式后台启动 EMQ X 时，控制台日志的副本文件。比如 erlang.log.1 , erlang.log.2 …
详情见官方文档-日志参数配置
3. ACL 发布订阅权限控制
发布订阅 ACL 指对 发布 (PUBLISH)/订阅 (SUBSCRIBE) 操作的 权限控制。例如拒绝用户名为 Anna 向 open/elsa/door 发布消息。
EMQ X 支持通过客户端发布订阅 ACL 进行客户端权限的管理，官网关于ACL详细介绍
HTTP 认证使用外部自建 HTTP 应用认证授权数据源，根据 HTTP API 返回的数据判定授权结果，能够实现复杂的 ACL 校验逻辑。
功能搭建
搭建实现过程
修改emqx配置文件. 要启用 HTTP ACL，需要在 etc/plugins/emqx_auth_http.conf 中配置superuser 请求
##-------------------------------------------------------------------- ## Superuser request. ## ## Variables: ## - %u: username ## - %c: clientid ## - %a: ipaddress ## - %r: protocol ## - %P: password ## - %p: sockport of server accepted ## - %C: common name of client TLS cert ## - %d: subject of client TLS cert ## ## Value: URL auth.http.super_req = http://127.0.0.1:8991/mqtt/superuser ## Value: post | get | put auth.http.super_req.method = post ## Value: Params auth.http.super_req.params = clientid=%c,username=%uACL 授权查询请求
##-------------------------------------------------------------------- ## ACL request. ## ## Variables: ## - %A: 1 | 2, 1 = sub, 2 = pub ## - %u: username ## - %c: clientid ## - %a: ipaddress ## - %r: protocol ## - %m: mountpoint ## - %t: topic ## ## Value: URL auth.http.acl_req = http://127.0.0.1:8991/mqtt/acl ## Value: post | get | put auth.http.acl_req.method = post ## Value: Params auth.http.acl_req.params = access=%A,username=%u,clientid=%c,ipaddr=%a,topic=%t,mountpoint=%m
注意:
- 配置文件配置好后, 需要重启emqx
auth.http.super_req和auth.http.acl_req配置的是客户端应用程序superuser 请求和ACL 授权查询请求接口地址, 因此我们不仅要关注接口的地址正确与否, 还要编写这两个接口的代码
在原有的项目 emq-demo 中我们已经开发了基于HTTP API的认证Controller： AuthController ，按照我们的请求URL配置，我们需要在该Controller中添加两个接口方法，一个是用于查询superuser的，一个是用于进行ACL授权查询的，这两个方法分别如下：
//客户端代码地址 https://gitee.com/TimePause/emqxdemo/*** 查询客户端是否为超级用户* @param clientid* @param username* @return*/@PostMapping("/superuser")public ResponseEntity<?> superUser(@RequestParam("clientid") String clientid,@RequestParam("username") String username) {log.info("emqx 查询是否是超级用户,clientid={},username={}", clientid, username);if (clientid.contains("admin") || username.contains("admin")) {log.info("用户{}是超级用户", username); //是超级用户return new ResponseEntity<Object>(HttpStatus.OK);} else {log.info("用户{}不是超级用户", username); //不是超级用户return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);}}/*** 在这个方法中我们设置了：* 只有用户名为 emq-client2 的客户端能够去订阅 testtopic/# 的权限其他客户端都不可以* 只有用户名为 emq-client3 的客户端能够向 testtopic/123 发布消息的权限其他都没有* @param access 1: 订阅. 2:发布* @param username* @param clientid* @param ipaddr* @param topic* @param mountpoint* @return*/@PostMapping("/acl")public ResponseEntity acl(@RequestParam("access") int access,@RequestParam("username") String username,@RequestParam("clientid") String clientid,@RequestParam("ipaddr") String ipaddr,@RequestParam("topic") String topic,@RequestParam("mountpoint") String mountpoint) {log.info("EMQX发起客户端操作授权查询请求,access={},username={},clientid={},ipaddr= {},topic={},mountpoint={}", access, username, clientid, ipaddr, topic, mountpoint);if (username.equals("emq-client2") && topic.equals("testtopic/#") && access == 1) {log.info("客户端{}有权限订阅{}", username, topic);/*** EMQ X 在设备发布、订阅事件中使用当前客户端相关信息作为参数，向用户自定义的认证服务发起请求权限，* 通过返回的 HTTP 响应状态码 (HTTP statusCode) 来处理 ACL 授权请求。* 无权限：API 返回 4xx 状态码* 授权成功：API 返回 200 状态码* 忽略授权：API 返回 200 状态码且消息体 ignore*/return new ResponseEntity<>(HttpStatus.OK);}if (username.equals("emq-client3") && topic.equals("testtopic/123") && access == 2) {log.info("客户端{}有权限向{}发布消息", username, topic);return new ResponseEntity<>(null, HttpStatus.OK);}log.info("客户端{},username={},没有权限对主题{}进行{}操作", clientid, username, topic, access == 1 ? "订阅" : "发布");return new ResponseEntity(HttpStatus.UNAUTHORIZED);//无权限}emqx DashBoard 配置 (访问 http://emqx所在ip:18083/#/plugins)
在Dashboard中打开插件 emqx_auth_http ，其他的认证插件
emqx_auth_clientid , emqx_auth_username 可以暂时停掉
测试结果
通过MQTTX测试发布订阅权限是否成功
MQTTX安装教程在第三章第2节认证操作案例之username认证第5步
测试超级用户
连接参数, 账号admin,密码admin
订阅主题
向指定主题发送消息并通过订阅消息接收
因为是超级用户, 所以订阅和接收都没有限制
客户端应用程序输出信息
测试emq-client2(账号: emq-client2, 密码123456)
因为上面应用程序代表逻辑中规定, 只有用户名为 emq-client2 的客户端能够去订阅 testtopic/# 的权限其他客户端都不可以. 因此我们先测试emq-client2然后测试emq-client3能否订阅该主题
主题订阅
客户端应用程序输出信息
测试emq-client3 (账号: emq-client3, 密码: 123456)
因为客户端代码逻辑规定, 只有用户名为 emq-client3 的客户端能够向 testtopic/123 发布消息的权限其他都没有. 因此可以先去测试emq-client3能否订阅testtopic/# 后, 在测试其能否向testtopic/123 发布消息
连接参数
测试能否订阅 testtopic/# 主题
测试能否向 testtopic/123 主题发送消息(可以)
同时可以看到emq-client2接收到了emq-client3发送的消息
测试过程客户端应用程序输出
由测试结果123可以看出:
- 超级用户订阅和发布不会受到限制
- 通过客户端代码可以控制具体哪个用户能订阅到那个主题, 或者能够向哪个主题发布消息
4. WebHook插件
插件可以将所有 EMQ X 的事件及消息都发送到指定的 HTTP 服务器
只需要在etc/plugins/emqx_web_hook.conf 中配置用于通知emqx 客户端服务端消息事件的事件类型以及被通知的Web Server地址
插件原理
WebHook 的内部实现是基于钩子，但它更靠近顶层一些。它通过在钩子上的挂载回调函数，获取到 EMQ X中的各种事件，并转发至 emqx_web_hook 中配置的 Web 服务器。
以客户端成功接入(client.connected) 事件为例，其事件的传递流程如下：
模拟实现
配置 WebHook 插件
etc/plugins/emqx_web_hook.conf, 然后重启emqx## 回调的 Web Server 地址 web.hook.api.url = http://127.0.0.1:8080## 编码 Payload 字段 ## 枚举值: undefined | base64 | base62 ## 默认值: undefined (不进行编码) ## web.hook.encode_payload = base64## 通知Web Server消息、事件配置 web.hook.rule.client.connected.1 = {"action": "on_client_connected"} web.hook.rule.client.disconnected.1 = {"action": "on_client_disconnected"} web.hook.rule.client.subscribe.1 = {"action": "on_client_subscribe"} web.hook.rule.client.unsubscribe.1 = {"action": "on_client_unsubscribe"} web.hook.rule.session.created.1 = {"action": "on_session_created"} web.hook.rule.session.subscribed.1 = {"action": "on_session_subscribed"} web.hook.rule.session.unsubscribed.1 = {"action": "on_session_unsubscribed"} web.hook.rule.session.terminated.1 = {"action": "on_session_terminated"} web.hook.rule.message.publish.1 = {"action": "on_message_publish"} web.hook.rule.message.deliver.1 = {"action": "on_message_deliver"} web.hook.rule.message.acked.1 = {"action": "on_message_acked"}在通知Web Server消息、事件配置, 每个action对应的行为如下
编写回调的 Web Server地址, 依然是使用原有的项目 emq-demo
在原有的项目 emq-demo 中我们已经开发了基于HTTP API的认证Controller： AuthController , 现在我们将新建一个controller用于接收emqx通知以及获取所有客户端连接情况//客户端代码地址 https://gitee.com/TimePause/emqxdemo/*** webhook用于监听客户端连接情况** @author caohaiyang*/ @RestController @RequestMapping("/webhook") public class WebHookController {private static final Logger log = LoggerFactory.getLogger(WebHookController.class);private Map<String, Boolean> clientStatusMap = new HashMap<>();/*** 接收emqx通知事件结果* @param params*/@PostMapping("/hook")public void hook(@RequestBody Map<String, Object> params) {log.info("emqx 触发 webhook,请求体数据={}", params);String action = (String) params.get("action");String clientId = (String) params.get("clientid");if (action.equals("client_connected")) {//客户端成功接入clientStatusMap.put(clientId, true);}if (action.equals("client_disconnected")) {//客户端断开连接clientStatusMap.put(clientId, false);}}/*** 获取所有客户端连接情况* @return*/@GetMapping("/getall")public Map getAllStatus(){return clientStatusMap;} }测试准备, 需要在插件中开启webhook插件
http://emex所在服务器ip:18083/#/plugins进行测试
通过MQTTX登录任意一个任意一个用户, 可以看到相关日志显示客户端在连接中有三个过程: 1. 客户端发起连接 2.服务端下发连接应答 3. 客户端成功接入
浏览器访问第二个接口, 获取所有客户端连接情况http://emqx所在ip:8991/webhook/getall ## mqttx_408f24bc 表示的是client id, 用于唯一标识该连接
5. 管理监控API的使用
EMQ X 提供了 HTTP API 以实现与外部系统的集成，例如查询客户端信息、发布消息和创建规则等。
EMQ X 的 HTTP API 服务默认监听 8081 端口，可通过 etc/plugins/emqx_management.conf 配置文件修改监听端口，或启用 HTTPS 监听。EMQ X 4.0.0 以后的所有 API 调用均以 api/v4 开头。
接口安全及响应码
EMQ X 的 HTTP API 使用 Basic 认证方式， id 和 password 须分别填写 AppID 和 AppSecret。
默认的AppID 和 AppSecret 是：amdin/public。你可以在 Dashboard 的左侧菜单栏里，选择 “MANAGEMENT” ->“Applications” 来修改和添加 AppID/AppSecret。
- 响应码
EMQ X 接口在调用成功时总是通过HTTP status code返回响应码
- 返回码
EMQ X 接口的响应消息体为 JSON 格式，其中总是包含返回码 code 。
接口请求工具
在第三章第2节 认证操作案例->client id认证第4步中, 我们已经使用了vscode的rest client 插件发送http请求. 下面我们将继续使用该插件来请求其他api
@hostname = emqx所在服务器ip
@port=8081
@contentType=application/json
@userName=admin
@password=public#############获取所有支持的API接口########
GET http://{{hostname}}:{{port}}/api/v4 HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}#############获取所有Broker基本信息########
GET http://{{hostname}}:{{port}}/api/v4/brokers HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}#############获取Broker基本信息########
GET http://{{hostname}}:{{port}}/api/v4/brokers/{node} HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}#############获取客户端列表信息########
GET http://{{hostname}}:{{port}}/api/v4/clients HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}#############获取集群下所有订阅信息########
GET http://{{hostname}}:{{port}}/api/v4/subscriptions HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}#############获取所有状态数据########
GET http://{{hostname}}:{{port}}/api/v4/stats HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}#############获取集群下当前告警信息########
GET http://{{hostname}}:{{port}}/api/v4/alarms/present HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}#############获取黑名单信息########
GET http://{{hostname}}:{{port}}/api/v4/banned HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}
6. 保留消息
当客户端建立订阅时，如果服务端存在主题匹配的保留消息，则这些保留消息将被立即发送给该客户端。借助保留消息，新的订阅者能够立即获取最近的状态，而不需要等待无法预期的时间，这在很多场景下非常重要的。
EMQ X 默认开启保留消息的功能，可以在 etc/emqx.conf 中修改 mqtt.retain_available 为 false 以禁用保留消息功能。如果 EMQ X 在保留消息功能被禁用的情况下依然收到了保留消息，那么将返回原因码为0x9A（不支持保留消息）的 DISCONNECT 报文。
应用场景
某车联网项目，车辆出租公司会实时监控所有车辆的GPS地理位置信息，这些信息是通过每个车辆每10分钟定时上报的GPS信息，这些信息需要展示在某调度系统的大屏上，该调度系统因为其他模块升级需要重新部署，升级后也需要去订阅获取所有车辆的GPS信息，上线完成后刚好错过了车辆最近一次上报的GPS信息，如果这些消息不是保留消息，该调度系统大屏上是空白的，必须等10分钟后才能调度这些车辆，10分钟内无法做出任何操作，用户体验非常差，但是如果这些信息是保留消息，该系统上线后立即就会收到最近所有车辆的位置信息，立即就可以展示然后进行调度。
保留消息配置
EMQ X 的保留消息功能是由 emqx_retainer 插件实现，该插件默认开启，通过修改 emqx_retainer 插件的配置，可以调整 EMQ X 储存保留消息的位置，限制接收保留消息数量和 Payload 最大长度，以及调整保留消息的过期时间。
emqx_retainer 插件默认开启，插件的配置路径为 etc/plugins/emqx_retainer.conf 。
## retained 消息存储方式
## - ram: 仅内存
## - disc: 内存和磁盘
## - disc_only: 仅磁盘
retainer.storage_type = ram## 最大存储数 (0表示未限制)
retainer.max_retained_messages = 0## 单条最大可存储消息大小
retainer.max_payload_size = 1MB## 过期时间, 0 表示永不过期
## 单位: h 小时; m 分钟; s 秒。如 60m 表示 60 分钟
retainer.expiry_interval = 0测试保留消息
可以通过DashBoard -> 工具 -> websocket 来模拟保留消息的实现以及效果
1.首先去新建一个保留消息(图1), 然后订阅. 查看接收到的消息(图2)
2.取消步骤1订阅(图3), 然后创建一个非保留消息(图4), 然后再去订阅并查看收到的消息(图5)
图1
图2
图3
图4
图5
7. 共享订阅
共享订阅是在多个订阅者之间实现负载均衡的订阅方式
共享订阅的主题格式是针对订阅端来指定的，例如： $share/g/t/a ；而消息的发布方是向主题： t/a发布消息。这样在订阅方才能达到负载均衡的效果
带群组的共享订阅
以
$share/<group-name>为前缀的共享订阅是带群组的共享订阅
group-name 可以为任意字符串，属于同一个群组内部的订阅者将以负载均衡接收消息，但 EMQ X 会向不同群组广播消息。
例如，假设订阅者 s1，s2，s3 属于群组 g1，订阅者 s4，s5 属于群组 g2。那么当 EMQ X 向这个主题发布消息 msg1 的时候：EMQ X 会向两个群组
- g1 和 g2 同时发送 msg1s1，s2，s3 中只有一个会收到 msg1
- s4，s5 中只有一个会收到 msg1
实现过程
需要修改之前acl验证代码, 将emq-client2, emq-client3 用户的订阅和权限放开
@PostMapping("/acl")public ResponseEntity acl(@RequestParam("access") int access,@RequestParam("username") String username,@RequestParam("clientid") String clientid,@RequestParam("ipaddr") String ipaddr,@RequestParam("topic") String topic,@RequestParam("mountpoint") String mountpoint) {log.info("EMQX发起客户端操作授权查询请求,access={},username={},clientid={},ipaddr= {},topic={},mountpoint={}", access, username, clientid, ipaddr, topic, mountpoint);if (username.equals("emq-client2") && topic.equals("testtopic/#") && access == 1) {log.info("客户端{}有权限订阅{}", username, topic);/*** EMQ X 在设备发布、订阅事件中使用当前客户端相关信息作为参数，向用户自定义的认证服务发起请求权限，* 通过返回的 HTTP 响应状态码 (HTTP statusCode) 来处理 ACL 授权请求。* 无权限：API 返回 4xx 状态码* 授权成功：API 返回 200 状态码* 忽略授权：API 返回 200 状态码且消息体 ignore*/return new ResponseEntity<>(HttpStatus.OK);}if (username.equals("emq-client3") && topic.equals("testtopic/123") && access == 2) {log.info("客户端{}有权限向{}发布消息", username, topic);return new ResponseEntity<>(null, HttpStatus.OK);}log.info("客户端{},username={},没有权限对主题{}进行{}操作", clientid, username, topic, access == 1 ? "订阅" : "发布");return new ResponseEntity(HttpStatus.OK);//无权限UNAUTHORIZED, 这里作用设置成共享订阅测试用户才设置成ok}需要额外开启的插件
模拟测试, 使用MQTTX 使用 emq-client2, emq-client3用户建立连接, 且都订阅
$share/g1/t1/a主题.登录emqx的Dashboard的websocket插件, 创建连接, 使用上面emqx-demo项目中创建的超级用户 admin/admin , 向
t1/a主题发布5条消息共享群组订阅消息.可以看到client2和client3随机收到了消息
不带群组的共享订阅消息
以 $queue/ 为前缀的共享订阅是不带群组的共享订阅。它是 $share 订阅的一种特例，相当与所有订阅者都在一个订阅组里面：
实现过程
基于上面带群组的共享订阅进行测试
使用MQTTX 使用 emq-client2, emq-client3用户建立连接, 且都订阅
$queue/g1/t1/a主题.登录emqx的Dashboard的websocket插件, 创建连接, 使用上面emqx-demo项目中创建的超级用户 admin/admin , 向
t1/a主题发布5条消息不带群组的共享群组订阅消息.可以看到client2和client3随机收到了消息
均衡策略与派发 Ack 配置
常用的均衡策略有随机, 轮询, 哈希等, 具体如下图
配置均衡策略
在emqx根目录下. 进入 /etc/emqx.conf , 通过修改broker.shared_dispatch_ack_enabled 属性来修改负载均衡策略
# 均衡策略
## Dispatch strategy for shared subscription
#### Value: Enum
## - random
## - round_robin
## - sticky
## - hash broker.shared_subscription_strategy = random
# 共享分发时是否需要 ACK，适用于 QoS1 QoS2 消息，启用时，当通过shared_subscription_strategy选中的 一个订阅者离线时，应该允许将消息发送到组中的另一个订阅者
broker.shared_dispatch_ack_enabled = false
- 延时发布
EMQ X 的延迟发布功能可以实现按照用户配置的时间间隔延迟发布 PUBLISH 报文的功能。当客户端使用特殊主题前缀$delayed/{DelayInteval} 发布消息到 EMQ X 时，将触发延迟发布功能。延迟发布的功能是针对消息发布者而言的，订阅方只需要按照正常的主题订阅即可。
应用场景
某智能售货机平台在双十一当天要对设备中所有商品做5折销售，双十一过去之后要立马恢复原价，为了满足这样的场景，我们可以在双十一0点给所有设备发送两条消息，一条消息是通过正常的主题发送，消息内容打5折；第二条消息延迟消息，延迟24小时，消息内容是恢复原价。这样在一个实现中可以完成两个业务场景。
消息发布格式
# DelayInterval延时时间, 单位秒, TopicName: 主题名
$delayed/{DelayInterval}/{TopicName}
功能实现
登录Dashboard
http://emqx所在ip:18083/#/plugins, 开启延时发布插件emqx_delayed_publish在websocket
http://emqx所在ip:18083/#/websocket中, 在创建连接后, 首先订阅一个主题t1/1, 然后根据这个主题按照延时发布的格式$delayed/10/t1/1发布消息{ "msg": "测试延时发布" }, 即: 消息发送后消息订阅者延时10s才能接收到消息
10. 代理订阅
静态/内置代理订阅
EMQ X 的代理订阅功能使得客户端在连接建立时，不需要发送额外的 SUBSCRIBE 报文，便能自动建立用户预设的订阅关系。
静态代理订阅的核心就是通过配置文件来配置订阅的主题, 在我们建立连接后就会自动为我们创建创建订阅
功能实现
代理订阅功能默认关闭，开启此功能需要修改 etc/emqx.conf 文件中的 module.subscription 配置项。默认 off 表示关闭，如需开启请修改为 on 。最后两行则是设置订阅的主题和Qos, 而且可以配置多个订阅主题,通过
module.subscription.{order}来区分第几个代理订阅的主题. 需要重启emqx配置才能生效!##-------------------------------------------------------------------- ## Subscription Module## Enable Subscription Module. ## ## Value: on | off module.subscription = on module.subscription.1.topic = testtopic/# module.subscription.1.qos = 2在websocket
http://emqx所在ip:18083/#/websocket中, 在创建连接后, 向testtopic/666发送消息{ "msg": "测试静态代理订阅" }, 然后无需订阅直接发送. 可以看到 即使用户没有订阅, 订阅消息列表中也显示了消息. 因为上面配置的代理订阅主题是testtopic/#
基于Webhook和API实现动态代理订阅
开启了 emqx_web_hook 组件后，EMQ X的事件都会勾起对我们配置的webhook接口进行回调，在该webhook接口中我们能够获取客户端的相关信息比如 clientId,username 等，然后我们可以在该接口方法中针对该客户端自动订阅某一主题，订阅的实现我们基于EMQ X给我们提供的监控管理的相关HTTP API，意味着我们调用相关的HTTP API可完成客户端订阅的功能，相关的HTTP API可在Dashboard中查看，也可以在官方的产品文档中查找：
功能概括就是: 让客户端连接时自动订阅某一主题, 反之客户端下线时我们可以自动取消订阅
功能实现
- 登录Dashboard
http://emqx所在ip:18083/#/plugins, 开启webhook插件emqx_web_hook
在原有的项目 emq-demo的WebHookController 中继续开发
业务逻辑是, 在客户端处于连接状态时(client_connected),调用自动订阅主题autoSub/#, 在断开连接状态时(client_disconnected)取消订阅该方法//客户端代码地址 https://gitee.com/TimePause/emqxdemo/*** 接收emqx配置的通知事件* @param params*/@PostMapping("/hook")public void hook(@RequestBody Map<String, Object> params) {log.info("emqx 触发 webhook,请求体数据={}", params);String action = (String) params.get("action");String clientId = (String) params.get("clientid");if (action.equals("client_connected")) {//客户端成功接入clientStatusMap.put(clientId, true);//自动订阅消息autoSub(clientId,"autoSub/#",QosEnum.Qos2,true);}if (action.equals("client_disconnected")) {//客户端断开连接clientStatusMap.put(clientId, false);//自动取消订阅autoSub(clientId,"autoSub/#",QosEnum.Qos2,false);}}/*** 自动订阅或取消订阅的方法* @param clientId 客户端ip* @param topicFilter 主题名* @param qos 消息发布类型* @param sub 是否自动订阅? true:自动订阅, false: 取消订阅*/private void autoSub(String clientId, String topicFilter, QosEnum qos, boolean sub){// 构建基于basic认证的发起请求的restful请求模板RestTemplate restTemplate = new RestTemplateBuilder().basicAuthentication("admin", "public").defaultHeader(MediaType.APPLICATION_JSON_VALUE).build();//请求体参数Map param = new HashMap();param.put("clientid",clientId);param.put("qos",qos.value());param.put("topic",topicFilter);log.info("请求emq的相关参数:{}",param);//请求头参数HttpHeaders headers = new HttpHeaders();headers.setContentType(MediaType.APPLICATION_JSON);HttpEntity<Object> entity = new HttpEntity<Object>(param,headers);//自动订阅逻辑if (sub){//新创建一个线程的作用是异步new Thread(()->{ResponseEntity<String> responseEntity =restTemplate.postForEntity("http://emqx所在ip:8081/api/v4/mqtt/subscribe", entity, String.class);log.info("自动订阅的结果:{}",responseEntity.getBody());// 这里如果缺少return会导致无论如何都会走取消订阅!}).start();//注意return的位置, 不然会变成无论如何都会执行取消订阅操作return;}//自动取消订阅ResponseEntity<String> responseEntity =restTemplate.postForEntity("http://emqx所在ip:8081/api/v4/mqtt/unsubscribe", entity, String.class);log.info("自动取消订阅的结果:{}",responseEntity.getBody());}测试
使用MQTTX新建连接, 查看 emq-demo 项目日志(图1), 并且向autoSub/123发送消息, 测试在客户端启动时自动订阅的主题autoSub/#是否生效(图2), 然后断开MQTTX连接查看emq–demo日志(图3)
图1
图2
图3
11. 主题重写
EMQ X 的主题重写功能支持根据用户配置的规则在客户端订阅主题、发布消息、取消订阅的时候将 A 主题重写为 B 主题。
EMQ X 的保留消息和延迟发布可以与主题重写配合使用，例如，当用户想使用延迟发布功能，但不方便修改客户端发布的主题时，可以使用主题重写将相关主题重写为延迟发布的主题格式。
应用场景：
某共享单车平台A运营着大量的共享单车，每个单车上都装有一个物联网终端芯片，芯片上的程序是将一些数据通过mqtt协议上报到EMQ服务器；该公司某一天收购了另一家共享单车平台B，B平台下原有的单车也是通过mqtt上报消息数据，但是消息主题跟A平台的不一样，如果A平台想接入B平台的车上报的数据，我们就需要把B平台下所有车上芯片程序更改一下，这样虽然可行但是会耗费大量的人力物力成本，这时我们通过主题重写就可以实现B平台下所有单车数据的接收，几乎不需要编码，成本非常低。
每条主题重写规则的格式：
module.rewrite.rule.<number> = 主题过滤器 正则表达式 目标表达式
注意事项
EMQ X 的主题重写规则需要用户自行配置，用户可以自行添加多条主题重写规则，规则的数量没有限制，但由于任何携带主题的 MQTT 报文都需要匹配一遍重写规则，因此此功能在高吞吐场景下带来的性能损耗与规则数量是成正比的，用户需要谨慎地使用此功能
在主题重写功能开启的前提下，EMQ X 在收到诸如 PUBLISH 报文等带有主题的 MQTT 报文时，将使用报文中的主题去依次匹配配置文件中规则的主题过滤器部分，一旦成功匹配，则使用正则表达式提取主题中的信息，然后替换至目标表达式以构成新的主题。
EMQ X 使用倒序读取配置文件中的重写规则，当一条主题可以同时匹配多条主题重写规则的主题过滤器时，EMQ X 仅会使用它匹配到的第一条规则进行重写，如果该条规则中的正则表达式与 MQTT 报文主题不匹配，则重写失败，不会再尝试使用其他的规则进行重写。因此用户在使用时需要谨慎的设计 MQTT 报文主题以及主题重写规则。
目标表达式中可以使用 $N 这种格式的变量匹配正则表达中提取出来的元素， $N 的值为正则表达式中提取出来的第 N 个元素，比如 $1 即为正则表达式提取的第一个元素。
正则表达式解析：
^匹配输入字符串的开始位置，除非在方括号表达式中使用，当该符号在方括号表达式中使用时，表示不接受该方括号表达式中的字符集合$匹配输入字符串的结尾位置( )表示一个标记一个子表达式的开始和结束位置，[标记一个中括号表达式的开始.匹配除换行符 \n 之外的任何单字符，+匹配前面的子表达式一次或多次*匹配前面的子表达式零次或多次?匹配前面的子表达式零次或一次|指明两项之间的一个选择{n} n是一个非负整数。匹配确定的 n 次{n,} n是一个非负整数。至少匹配n 次{n,m}m 和 n 均为非负整数，其中n <= m。最少匹配 n 次且最多匹配 m 次 \d 匹配一个数字字符。等价于 [0-9]
主题重写配置实例
module.rewrite.rule.1 = y/+/z/# ^y/(.+)/z/(.+)$ y/z/$2
module.rewrite.rule.2 = x/# ^x/y/(.+)$ z/y/x/$1
module.rewrite.rule.3 = x/y/+ ^x/y/(\d+)$ z/y/$1
配置解析
功能实现
- 主题重写功能默认关闭，开启此功能需要修改 etc/emqx.conf 文件中的 module.rewrite 配置项, 修改后重启 emqx
./emqx restart## Rewrite Module## Enable Rewrite Module. ## ## Value: on | off module.rewrite = on ##x/y/1 -> z/y/1 ,这里通过下面规则会将原来的主题x/y/1 重写从 z/y/1 module.rewrite.rule.1 = x/# ^x/y/(.+)$ z/y/$1 - 功能测试
登录MQTTX, 订阅重写后的主题, 然后向重写前的主题发送消息, 测试重写配置是否正确, 如下图
12. 黑名单配置
EMQ X 为用户提供了黑名单功能，用户可以通过相关的 HTTP API 将指定客户端加入黑名单以拒绝该客户端访问，除了客户端标识符以外，还支持直接封禁用户名甚至 IP 地址。
黑名单只适用于少量客户端封禁需求，如果有大量客户端需要认证管理，我们需要使用认证功能来实现。
自动封禁
在黑名单功能的基础上，EMQ X 支持自动封禁那些被检测到短时间内频繁登录的客户端，并且在一段时间内拒绝这些客户端的登录，以避免此类客户端过多占用服务器资源而影响其他客户端的正常使用。
需要注意的是，自动封禁功能只封禁客户端标识符，并不封禁用户名和 IP 地址，即该机器只要更换客户端标识符就能够继续登录。
自动封禁功能配置
此功能默认关闭，用户可以在 emqx.conf 配置文件中将 enable_flapping_detect 配置项设为 on 以启用此功能。
# 开启自动封禁检测 zone.external.enable_flapping_detect = on # 用户可以在 emqx.conf 配置文件中调整触发阈值和封禁时长等配置 flapping_detect_policy = 30, 1m, 5m # 此配置项的值以 , 分隔，依次表示客户端离线次数，检测的时间范围以及封禁时长， # 因此上述默认配置即表示如果客户端在 1 分钟内离线次数达到 30 次，那么该客户端使用的客户端标识符将被封禁 5 分钟。
手动配置(通过管理端api实现)
添加黑名单配置参数
在vscode通过rest client 插件, 创建http格式文件, 发送http请求
@hostname = emqx所在ip @port=8081 @contentType=application/json @userName=admin @password=public######获取黑名单列表####### GET http://{{hostname}}:{{port}}/api/v4/banned HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}}######添加黑名单####### POST http://{{hostname}}:{{port}}/api/v4/banned HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}} { "who": "emq-client2", "as": "username" }#########删除黑名单############### DELETE http://{{hostname}}:{{port}}/api/v4/banned/username/emq-client2 HTTP/1.1 Content-Type: {{contentType}} Authorization: Basic {{userName}}:{{password}}
13. 速率限制
EMQ X 提供对接入速度、消息速度的限制：当客户端连接请求速度超过指定限制的时候，暂停新连接的建立；当消息接收速度超过指定限制的时候，暂停接收消息。
速率限制原理
EMQ X 使用令牌桶Token Bucket算法来对所有的 Rate Limit 来做控制。令牌桶算法 的逻辑如下图:
- 存在一个可容纳令牌(Token) 的最大值 burst 的桶(Bucket)，最大值 burst 简记为 b 。
- 存在一个 rate 为每秒向桶添加令牌的速率，简记为 r 。当桶满时则不不再向桶中加入令牌。
- 每当有 1 个(或 N 个)请求抵达时，则从桶中拿出 1 个 (或 N 个) 令牌。如果令牌不不够则阻塞，等待令牌的生成。
由此可知该算法中:
长期来看，所限制的请求速率的平均值等于 rate 的值。
记实际请求达到速度为 M，且 M > r，那么，实际运行中能达到的最大(峰值)速率为 M = b + r. 证明: 最大速率 M 为：能在1个单位时间内消耗完满状态令牌桶的速度。而桶中令牌的消耗速度为M - r，故可知：b / (M - r) = 1，得 M = b + r
功能实现
- 速率限制是一种 backpressure 方案，从入口处避免了系统过载，保证了系统的稳定和可预测的吞吐。速率限制可在 etc/emqx.conf 中配置：
max_conn_rate 是单个 emqx 节点上连接建立的速度限制。 1000 代表每秒最多允许 1000 个客户端接入。
publish_limit 是单个连接上接收 PUBLISH 报文的速率限制。 100,10s 代表每个连接上允许收到的最大PUBLISH 消息速率是每 10 秒 100 个。
rate_limit 是单个连接上接收 TCP数据包的速率限制。 100KB,10s 代表每个连接上允许收到的最大 TCP报文速率是每 10 秒 100KB。 publish_limit 和 rate_limit 提供的都是针对单个连接的限制，EMQ X 目前没有提供全局的消息速率限制。
当使用如下配置做报文速率限制的时候：
listener.tcp.external.rate_limit = 100KB,10sEMQ X 将使用两个值初始化每个连接的 rate-limit 处理器：
rate = 100 KB / 10s = 10240 B/s burst = 100 KB = 102400 B根据消息速率限制原理中的算法，可知：
# 根据令牌桶算法 长期来看允许的平均速率限制为 10240 B/s (平均速率r) 允许的峰值速率为 102400 + 10240 = 112640 B/s (M=b+r)为提高系统吞吐，EMQ X 的接入模块不会一条一条的从 socket 读取报文，而是每次从 socket 读取 N 条报文。rate-limit 检查的时机就是在收到这 N 条报文之后，准备继续收取下个 N 条报文之前。故实际的限制速率不会如算法一样精准。EMQ X 只提供了一个大概的速率限制。 N 的值可以在 etc/emqx.conf 中配置：
14. 消息重传
消息重传 (Message Retransmission) 是属于 MQTT 协议标准规范的一部分。
协议中规定了作为通信的双方 服务端 和 客户端 对于自己发送到对端的 PUBLISH 消息都应满足其 服务质量(Quality of Service levels) 的要求。如：
- QoS 1：表示 消息至少送达一次 (At least once delivery)；即发送端会一直重发该消息，除非收到了对端对该消息的确认。在 MQTT 协议的上层（即业务的应用层）相同的 QoS 1 消息可能会收到多次。
- QoS 2：表示 消息只送达一次 (Exactly once delivery)；即该消息在上层仅会接收到一次。
虽然，QoS 1 和 QoS 2 的 PUBLISH 报文在 MQTT 协议栈这一层都会发生重传，但注意：
- QoS 1 消息发生重传后，在 MQTT 协议栈上层，也会收到这些重发的 PUBLISH 消息。
- QoS 2 消息无论如何重传，最终在 MQTT 协议栈上层，都只会收到一条 PUBLISH 消息
功能实现
有两种场景会导致消息重发：
- PUBLISH 报文发送给对端后，规定时间内未收到应答。则重发这个报文。
- 在保持会话的情况下，客户端重连后；EMQ X 会自动重发 未应答的消息，以确保 QoS 流程的正确。
在 etc/emqx.conf 中可配置：
15. 系统调优
EMQ X 消息服务器 4.x 版本 MQTT 连接压力测试到 130 万，在一台 8 核心、32G 内存的 CentOS 服务器上。100 万连接测试所需的 Linux 内核参数，网络协议栈参数，Erlang 虚拟机参数.EMQ X 消息服务器参数以及测试客户端设置可见官网 系统调优 介绍, 非常全面
五. 规则引擎
介绍及原理
EMQ X Rule Engine (以下简称规则引擎) 用于配置 EMQ X 消息流与设备事件的处理、响应规则。
规则引擎用于配置一套规则，该规则是针对EMQ X的消息流和设备事件如何处理的一套细则。
原理
EMQ X 在 消息发布 或 事件触发 时将触发规则引擎，满足触发条件的规则将执行各自的 SQL 语句筛选并处理消息和事件的上下文信息。
- 消息发布
规则引擎借助响应动作可将特定主题的消息处理结果存储到关系型数据库（mysql，PostgreSQL），NoSql（Redis，MongoDB），发送到 HTTP Server，转发到消息队列 Kafka 或 RabbitMQ，重新发布到新的主题甚至是另一个 Broker 集群中，每个规则可以配置多个响应动作。
- 事件触发
规则引擎使用
$events/开头的虚拟主题（事件主题）处理 EMQ X 内置事件，内置事件提供更精细的消息控制和客户端动作处理能力，可用在 QoS 1 QoS 2 的消息抵达记录、设备上下线记录等业务中。
应用场景
- 动作监听：智慧家庭智能门锁开发中，门锁会因为网络、电源故障、人为破坏等原因离线导致功能异常，使用规则引擎配置监听离线事件向应用服务推送该故障信息，可以在接入层实现第一时间的故障检测的能力；
- 数据筛选：车辆网的卡车车队管理，车辆传感器采集并上报了大量运行数据，应用平台仅关注车速大于40 km/h 时的数据，此场景下可以使用规则引擎对消息进行条件过滤，向业务消息队列写入满足条件的数据；
- 消息路由：智能计费应用中，终端设备通过不同主题区分业务类型，可通过配置规则引擎将计费业务的消息接入计费消息队列并在消息抵达设备端后发送确认通知到业务系统，非计费信息接入其他消息队列，实现业务消息路由配置；
- 消息编解码：其他公共协议 / 私有 TCP 协议接入、工控行业等应用场景下，可以通过规则引擎的本地处理函数（可在 EMQ X 上定制开发）做二进制 / 特殊格式消息体的编解码工作；亦可通过规则引擎的消息路由将相关消息流向外部计算资源如函数计算进行处理（可由用户自行开发处理逻辑），将消息转为业务易于处理的 JSON 格式，简化项目集成难度、提升应用快速开发交付能力。
规则引擎组成
与 EMQ X 规则引擎相关的概念包括: 规则(rule)、动作(action)、资源(resource) 和 资源类型(resourcetype)。 规则、动作、资源的关系:
- 规则(Rule): 规则由 SQL 语句和动作列表组成。动作列表包含一个或多个动作及其参数。
- SQL 语句用于筛选或转换消息中的数据。
- 动作(Action) 是 SQL 语句匹配通过之后，所执行的任务。动作定义了一个针对数据的操作。 动作可以绑定资源，也可以不绑定。例如，“inspect” 动作不需要绑定资源，它只是简单打印数据内容和动作参数。而“data_to_webserver” 动作需要绑定一个 web_hook 类型的资源，此资源中配置了 URL。
- 资源(Resource): 资源是通过资源类型为模板实例化出来的对象，保存了与资源相关的配置(比如数据库连接地址和端口、用户名和密码等) 和系统资源(如文件句柄，连接套接字等)。
- 资源类型 (Resource Type): 资源类型是资源的静态定义，描述了此类型资源需要的配置项。
SQL语句
基本语法
FROM、SELECT 和 WHERE 子句:
SQL 语句用于从原始数据中，根据条件筛选出字段，并进行预处理和转换，基本格式为:SELECT <字段名> FROM <主题> [WHERE <条件>] FROM 子句将规则挂载到某个主题上（向该主题发布消息时触发，该主题是事件主题则事件发生时触发） SELECT 子句用于选择输出结果中的字段 WHERE 子句用于根据条件筛选消息FOREACH、DO 和 INCASE 子句:
如果对于一个数组数据，想针对数组中的每个元素分别执行一些操作并执行 Actions，需要使用 FOREACH-DO- INCASE 语法。其基本格式为:FOREACH <字段名> [DO <条件>] [INCASE <条件>] FROM <主题> [WHERE <条件>]FOREACH 子句用于选择需要做 foreach 操作的字段，注意选择出的字段必须为数组类型 DO 子句用于对 FOREACH 选择出来的数组中的每个元素进行变换，并选择出感兴趣的字段 INCASE 子句用于对 DO 选择出来的某个字段施加条件过滤其中 DO 和 INCASE 子句都是可选的。DO 相当于针对当前循环中对象的 SELECT 子句，而 INCASE 相当于针对当前循环中对象的 WHERE 语句。
基本语法举例
从 topic 为 “t/a” 的消息中提取所有字段
SELECT * FROM "t/a"从 topic 为 “t/a” 或 “t/b” 的消息中提取所有字段:
SELECT * FROM "t/a","t/b"从 topic 能够匹配到 ‘t/#’ 的消息中提取所有字段
SELECT * FROM "t/#"从 topic 能够匹配到 ‘t/#’ 的消息中提取 qos, username 和 clientid 字段
SELECT qos, username, clientid FROM "t/#"从任意 topic 的消息中提取 username 字段，并且筛选条件为 username = ‘Steven’:
SELECT username FROM "#" WHERE username='Steven'从任意 topic 的 JSON 消息体(payload) 中提取 x 字段，并创建别名 x 以便在 WHERE 子句中使用。WHERE 子句限定条件为 x = 1。下面这个 SQL 语句可以匹配到消息体 {“x”: 1}, 但不能匹配到消息体 {“x”: 2}:
SELECT payload as p FROM "#" WHERE p.x = 1类似于上面的 SQL 语句，但嵌套地提取消息体中的数据，下面的 SQL 语句可以匹配到 JSON 消息体 {“x”:{“y”: 1}}:
SELECT payload as a FROM "#" WHERE a.x.y = 1在 clientid = ‘c1’ 尝试连接时，提取其来源 IP 地址和端口号:
SELECT peername as ip_port FROM "$events/client_connected" WHERE clientid = 'c1'筛选所有订阅 ‘t/#’ 主题且订阅级别为 QoS1 的 clientid:
SELECT clientid FROM "$events/session_subscribed" WHERE topic = 't/#' and qos = 1筛选所有订阅主题能匹配到 ‘t/#’ 且订阅级别为 QoS1 的 clientid。注意与上例不同的是，这里用的是主题匹配操作符 ‘=~’，所以会匹配订阅 ‘t’ 或 ‘t/+/a’ 的订阅事件:
SELECT clientid FROM "$events/session_subscribed" WHERE topic =~ 't/#' and qos = 1FROM 子句后面的主题需要用双引号 "" 引起来。 WHERE 子句后面接筛选条件，如果使用到字符串需要用单引号 '' 引起来。 FROM 子句里如有多个主题，需要用逗号 "," 分隔。例如 SELECT * FROM "t/1", "t/2" 。 可以使用使用 "." 符号对 payload 进行嵌套选择
CASE-WHEN 语法示例
将消息中 x 字段的值范围限定在 0~7 之间
SELECT CASE WHEN payload.x < 0 THEN 0 WHEN payload.x > 7 THEN 7 ELSE payload.x END as x FROM "t/#"假设消息为:
{"x": 9}, 则输出结果为 {“x”: 7}测试结果
进入Dashboard的规则页面,http://服务器所在ip:18083/#/rules/create, 则规则SQL中输入SQL, 并且开启测试. 会自动生成username, topic, qos. payload, client. 然后我们修改下payload, 改为{"msg": "hello","x":9}
运行测试功能吗可以看到结果为{ "x": 7}
更多语法规则以及规则SQL查询时可用字段介绍可见官网 SQL语句 介绍部分
规则引擎接收案例
现需要通过规则引擎提取出从 username=emq-client2 的客户端发送过来原始数据中的msg,user,orderNo 等数据，需要过滤 password 字段，同时还需要提取消息发布的qos信息，然后将最终过滤出来的消息通知到我们的web服务上。
功能实现
确认主题消息格式模板, 作为最后通过mqttx发送的消息内容
{ "msg": "hello", "user":"emq-client2", "password":"123456", "orderNo":"12345sfd" }创建资源
打开 emqx dashboard ，选择左侧的 “规则引擎” 选项卡http://emqx所在ip:18083/#/rules，打开资源页面. 点击新建，创建资源. 选择webhook
编辑webhook资源信息, 这里主要是编辑提供接收规则引擎处理后的消息的接口地址(目前还没编写该接口, 后续会创建), 测试连接后新建资源
新建规则(DashBoard->规则引擎->规则)
输入规则SQL, 然后在 响应动作 一栏添加响应动作SELECT payload.msg as msg, payload.user as user, payload.orderNo as orderNo, qos FROM"rule/#" WHERE username = 'emq-client2'
设置响应动作为发送数据到web服务, 然后关联资源, 所关联的资源就是我们在上一步创建的资源
创建资源webhook接口
修改原有代码(地址：https://gitee.com/TimePause/emqxdemo)，添加一个http接口用来接收emq通过Post发送过来的数据，在这里我们只是简单的输出到控制台，证明我们的程序已经接收到了通过自己创建的规则引擎转发过来的数据。在实际业务中，我们会将接收到的数据进行后续复杂的业务处理，@RestController @RequestMapping("/resource") public class RuleController {private static final Logger log = LoggerFactory.getLogger(RuleController.class);/*** 用于接收规则引擎sql处理过后的信息* http://127.0.0.1:8991/resource/message*/@PostMapping("/message")public void message(@RequestBody Map<String,Object> params){System.out.println("接收规则引擎信息");params.entrySet().stream().forEach(x->{System.out.println(x.getKey()+":"+x.getValue());});} }利用mqttx发送消息
在上面项目启动后, 可编辑mqttx去新建连接, 连接参数如下
向规则引擎下去发送消息, 需要注意的是规则引擎订阅的主题是rule/#且是系统配置, 因此我们只需要向符合规则引擎订阅的主题(例如:rule/123)发送消息即可. 发送的内容则是我们在第1步配置好的主题消息格式模板
查看项目是否打印我们通过配置 规则sql后得到的消息
现将该教程所使用的emq x,mqttx, xhell+xftp 通过百度云分享如下:
觉得不错麻烦动动手点个赞吧~~~
链接：https://pan.baidu.com/s/1dSgs79rw5bnSkQyH-FbR4A
提取码：6mte
复制这段内容后打开百度网盘手机App，操作更方便哦–来自百度网盘超级会员V4的分享
Java物联网开发(二) —— 开源百万级分布式 MQTT 消息服务器EMQX相关推荐
- 物联网设备数据流转之搭建环境：开源云原生分布式物联网MQTT消息服务器EMQX
背景 EMQX Broker 一开始的定位就是物联网消息中间件,目前开源版本功能已经非常强大,而企业版本与Cloud版本更是提供了高阶功能,全托管.更稳定.更可靠,技术支持更及时.开源版的 EMQX ...
- Java 快速开发二维码生成服务
点击上方蓝色"程序猿DD",选择"设为星标" 回复"资源"获取独家整理的学习资料! 来源 | 公众号「码农小胖哥」 1. 前言 不知道从什么 ...
- Java物联网开发(一) —— MQTT协议
基于 Publish/Subscribe 模式的物联网通信协议MQTT 一. 简介 二. 特点 QoS消息发布质量介绍 三. 应用场景 四. 原理 MQTT协议中的方法 五. MQTT协议数据包结构 ...
- 专为物联网开发的开源操作系统Contiki（转）
专为物联网开发的开源操作系统Contiki(转) (2012-04-19 15:31:09) 原文网址:http://blog.sina.com.cn/s/blog_6de000c201010z7n. ...
- 物联网实战-基于开源 MQTT消息服务器EMQ X
物联网协议对比 对于物联网,最重要的是在互联网中设备与设备的通讯,现在物联网在internet通信中比较常见的通讯协议包括:HTTP.websocket.XMPP.COAP.MQTT HTTP和web ...
- JAVA应用开发MQ实战最佳实践——Series2：消息队列RocketMQ性能测试案例
简介:JAVA应用开发MQ实战最佳实践--Series2:消息队列RocketMQ性能测试案例 往期内容 JAVA应用开发MQ实战最佳实践--Series1:RocketMQ综述及代码设计 1. 消息 ...
- 千万级车联网 MQTT 消息平台架构设计
在本专题系列文章中,我们将根据 EMQ 在车联网领域的实践经验,从协议选择等理论知识,到平台架构设计等实战操作,与大家分享如何搭建一个可靠.高效.符合行业场景需求的车联网平台. 前言 随着整个汽车出行 ...
- 日调用量超三十万亿，腾讯开源百万级服务发现和治理中心北极星
目前很多企业在微服务实施和演化过程中,都会面临技术栈的多样性问题.整个微服务领域逐渐沉淀出了无数个相关组件,大家在选择上更加困难,也为企业的基础设施建设不断带来挑战.腾讯也曾面临这样的痛点,因此从 2 ...
- 蘑菇街 java 面试_【蘑菇街java工程师面试】蘑菇街Java后台开发二面挫败-看准网...
面试官是个女生,hr打电话来两天后就开始电话面试了.自我介绍 :大三学生,学习了一年iOS,自己做了两个项目,天气和微博,swift和OC都写. 蘑菇街面经一面: 1,如何用实现微信两人之间发送消息? ...
最新文章
- matlab 2012 vs2010混合编程
- Invoke shell in Java
- Ubuntu安装MDK
- 前端学习 -- Css -- 属性选择器
- 详解CockroachDB事务处理系统
- 一个微软面试题--关于位结构体
- unity2d随机生成物体_平面测量路径生成的4种方式
- Animoca Brands 旗下子公司 GAMEE 将于 4 月 2 日启动公募
- c++ 使用nacos_超赞！用阿里开源的Nacos做SpringCloud注册中心真贴心...
- JDK8高性能队列“Disruptor“
- SSH框架下的表单重复提交
- linux下解压java压缩包报错
- .java和.class的区别_Java中class与Class有什么区别
- 常见（MySQL）面试题（含答案）
- 重庆理工大学c语言实验答案,重庆理工大学C语言实验答案.doc
- python实现归结演绎推理_人工智能——归结演绎推理
- 小工具：找出序列中的极值点
- java.sql.SQLException: Undefined Error
- Dfinity(ICP)基础开发教程-5
- python 拼音相似度_多种相似度计算的python实现