目录
- EMQX WebHook介绍
- EMQX WebHook是什么
- EMQX WebHook配置项
- 如何使用EMQX WebHook配置
- WebHook配置事件推送参数详解
- SpringBoot集成Webhook实现客户端断连监控
- 1. 实现前提
- 2. 代码实现接口
- 3. 监听结果
- 总结
EMQX WebHook介绍
EMQX WebHook是什么
EMQX WebHook 是由 emqx_web_hook (opens new window)插件提供的将EMQX中的钩子事件通知到某个Web服务的功能。
WebHook 的内部实现是基于钩子，借助 Webhook 可以完成设备在线、上下线记录，订阅与消息存储、消息送达确认等诸多业务。
它通过在钩子上的挂载回调函数，获取到 EMQX 中的各种事件，并转发至 emqx_web_hook 中配置的 Web 服务器。
以 客户端成功接入(client.connected) 事件为例，其事件的传递流程如下：
Client | EMQX | emqx_web_hook | HTTP +------------+=============>| - - - - - - -> - - - - - - - ->===========> | Web Server || Broker | | Request +------------+注意：WebHook 对于事件的处理是单向的，它仅支持将 EMQX 中的事件推送给 Web 服务，并不关心 Web 服务的返回。
EMQX WebHook配置项
Webhook 的配置文件位于 etc/plugins/emqx_web_hook.conf，配置项的详细说明可以查看官方文档配置
在默认etc/plugins/emqx_web_hook.conf中，官方为我们提供了如下的配置触发规则：
##====================================================================
## WebHook
##====================================================================web.hook.api.url = http://127.0.0.1:80## Format:
## web.hook.rule.<HookName>.<No> = <Spec>
#web.hook.rule.client.connect.1 = {"action": "on_client_connect"}
#web.hook.rule.client.connack.1 = {"action": "on_client_connack"}
#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.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.delivered.1 = {"action": "on_message_delivered"}
#web.hook.rule.message.acked.1 = {"action": "on_message_acked"}
其中webhook支持的事件如下：
如何使用EMQX WebHook配置
-
手动修改Webhook的配置文件位于：
etc/plugins/emqx_web_hook.conf.
修改配置非常简单，打开指定文件后，官方已经写好这些了，如上面所示，
只需要修改web.hook.url，和在你需要的钩子事件前的#号去掉即可 -
emqx 启用插件 webhook插件（emqx有web控制台，在插件里启用就可以）
刷新后emqx_web_hook插件运行中即启动成功 -
然后就是自行搭建一个web接口服务，当客户端有指定事件后会主动推送，
接口地址即为Webhook的配置文件中配置的web.hook.api.url
WebHook配置事件推送参数详解
-
client.connect（处理连接报文）
-
client.connack（下发连接应答）
-
client.connected（成功接入）
-
client.disconnected（连接断开）
-
client.subscribe（订阅主题）
-
client.unsubscribe（取消订阅）
-
session.subscribed（会话订阅主题）
同 client.subscribe，action 为 session_subscribed -
session.unsubscribed（会话取消订阅）
同 client.unsubscribe，action 为 session_unsubscribe -
message.publish（消息发布）
-
message.delivered（消息投递）
-
message.acked（消息回执）
SpringBoot集成Webhook实现客户端断连监控
通过使用Emqx的Webhook插件实现监控所有客户端的连接状态，并做出客户端上线和下线后的逻辑处理
1. 实现前提
修改 etc/plugins/emqx_web_hook.conf 文件，设置事件转发的url和地址和触发规则
# 事件需要转发的目的服务器地址
web.hook.api.url = http://127.0.0.1:8000/api/mqtt/webhook# 打开下面两个触发规则
web.hook.rule.client.connected.1 = {"action": "on_client_connected"}
web.hook.rule.client.disconnected.1 = {"action": "on_client_disconnected"}
2. 代码实现接口
通过EMQX 的webhook将客户端的连接断开等事件通知到我们自建的服务上，通过事件类型获取客户端的连接状态，然后将客户端的连接状态进行存储，并且提供HTTP API供后台系统查询所有客户端的状态。
package com.team.modules.mqtt.controller;import com.google.gson.Gson;
import com.team.modules.mqtt.client.MyMQTTClient;
import com.team.modules.mqtt.service.SenderService;
import com.team.modules.system.domain.vo.ResultVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;import java.util.HashMap;
import java.util.Map;@RestController
@Slf4j
@RequestMapping("/api/mqtt")
public class SenderController {@Autowiredprivate MyMQTTClient myMQTTClient;@PostMapping("/webhook")public void webhook(@RequestBody() Map<String, Object> message) {
// 监听连接的上线和下线System.out.println("收到的消息：" + message);String action = (String) message.get("action");String clientid = (String) message.get("clientid");if (action.equals("client_connected")) {log.info("client:{} 上线", clientid);}if (action.equals("client_disconnected")) {log.info("client:{} 下线", clientid);}}
}
3. 监听结果
启动服务后，模拟客户端连接emqx，查看服务记录的日志：
总结
由此看出客户端建立连接和断开连接后，Emqx服务都会给配置的接口web.hook.api.url = http://127.0.0.1:8000/api/mqtt/webhook发送一条数据，该数据字段可以参考上述的：WebHook配置事件推送参数详解。使用此方法可以将mqtt应用的更加灵活。其他的监听事件配置也类似，遇到合适场景可以自行发挥。
