EMQX监听客户端上下线的 2 种方式

需求说明：设备客户端连上emqx时，后台显示设备为在线状态；设备客户端断开emqx时，后台显示设备为离线状态。 下面介绍两种监听客户端上下线的思路。

方案一：订阅$SYS系统主题

提示：EMQX 默认只允许本机的 MQTT 客户端订阅 $SYS 主题，请参照 内置 ACL 修改发布订阅 ACL 规则。

• 监听客户端上下线事件

$SYS` 主题前缀: `$SYS/brokers/${node}/clients/

• connected 事件消息的 Payload 解析成 JSON 格式如下:

{
  "username": "foo",
  "ts": 1625572213873,
  "sockport": 1883,
  "proto_ver": 4,
  "proto_name": "MQTT",
  "keepalive": 60,
  "ipaddress": "127.0.0.1",
  "expiry_interval": 0,
  "connected_at": 1625572213873,
  "connack": 0,
  "clientid": "emqtt-8348fe27a87976ad4db3",
  "clean_start": true
}

• disconnected 事件消息的 Payload 解析成 JSON 格式如下:

{
  "username": "foo",
  "ts": 1625572213873,
  "sockport": 1883,
  "reason": "tcp_closed",
  "proto_ver": 4,
  "proto_name": "MQTT",
  "ipaddress": "127.0.0.1",
  "disconnected_at": 1625572213873,
  "clientid": "emqtt-8348fe27a87976ad4db3"
}

思路：

1、订阅主题$SYS/brokers/${node}/clients/${clientid}/(connected/disconnected)

2、订阅后客户端上下线均会发送上下线消息，然后在接收消息端处理业务逻辑即可。

缺点：

1、该方案因为是订阅主题，如果emqx服务崩溃导致重连，订阅该主题的客户端也需要重连，在重连期间或者其它客户端已经连接，会造成没能收到主题消息，可靠性差。

2、实际生产中没法使用，而且官方也不推荐这种方式。 （项目最初用的是这种方式，因为简单，但后面出现客户端在线状态判断不准确问题。）

方案二：EMQX WebHook plugin监听

2.1 WebHook介绍 

WebHook的功能，用一句话概括就是，当客户端上下线后，给指定的Web地址发送一条消息。

• WebHook是由GitHub - emqx/emqx-web-hook: EMQX Webhook Plugin 插件提供的将EMQX中的钩子事件通知到某个Web服务的功能。
• WebHook 的内部实现是基于钩子，但它更靠近顶层。它通过在钩子上的挂载回调函数，获取到 EMQX 中的各种事件，并转发至 emqx_web_hook 中配置的 Web 服务器。
• WebHook 对于事件的处理是单向的，它仅支持将 EMQX 中的事件推送给 Web 服务，并不关心 Web 服务的返回。借助 Webhook 可以完成设备在线、上下线记录，订阅与消息存储、消息送达确认等诸多业务。

2.2 WebHook的使用 

• Webhook 的配置文件位于 etc/plugins/emqx_web_hook.conf，配置项的详细说明可以查看 配置项。这个配置非常简单，打开指定文件后，官方已经写好这些了，只需要修改web.hook.url，和在你需要的钩子事件前的#号去掉即可。

web.hook.url = http://127.0.0.1:8080/webhook

web.hook.rule.client.connected.1     = {"action": "on_client_connected"}
web.hook.rule.client.disconnected.1  = {"action": "on_client_disconnected"}
#web.hook.rule.client.connect.1       = {"action": "on_client_connect"}
#web.hook.rule.client.connack.1       = {"action": "on_client_connack"}
#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"}

• emqx 启用插件 webhook插件（emqx有web控制台，在插件里启用就可以）

• 然后就是自行搭建一个web服务，当客户端有指定事件后会主动推送。

优点：

1、该方案只要web服务端不崩溃，就一定会收到上下线事件，可靠性有保障。

2、耦合性更低。

备注：EMQX5.0以后的版本，提供了Webhook配置的图形化界面，配置更加友好。

Webhook | EMQX 5.3 文档