Skip to content

事件订阅

EMQX Platform 数据集成提供了以下七个可用的事件订阅主题。

事件主题名释义
$events/client_connected设备上线
$events/client_disconnected设备下线
$events/message_delivered消息投递
$events/message_acked消息确认
$events/message_dropped消息在转发的过程中被丢弃
$events/delivery_dropped消息在投递的过程中被丢弃
$events/session_subscribed设备订阅
$events/session_unsubscribed设备取消订阅

本文将从设备上下线、消息通知和主题订阅通知,三个方面分别介绍事件订阅主题使用和各个字段的含义。

设备上下线通知

设备上线

设备上线的消息主题为 $events/client_connected。在规则创建页面中输入如下规则匹配 SQL 语句便可完成设备上线规则的创建,当新设备上线并连接到 EMQX Cloud 便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/client_connected"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 设备上线

各字段释义如下。

字段释义
clean_startMQTT clean_start
clientid客户端 ID
connected_at终端连接完成时间 (s)
event事件类型,固定为 "client.connected"
expiry_intervalMQTT Session 过期时间
is_bridge是否为 MQTT bridge 连接
keepaliveMQTT 保活间隔
mountpoint主题挂载点(主题前缀)
node事件触发所在节点
peername终端的 IPAddress 和 Port
proto_name协议名字
proto_ver协议版本
socknameemqx 监听的 IPAddress 和 Port
timestamp事件触发时间 (ms)
username用户名

设备下线

设备下线的消息主题为 $events/client_disconnected。在规则创建页面中输入如下规则匹配 SQL 语句便可完成设备下线规则的创建,当在线设备下线断开连接时便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/client_disconnected"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 设备下线

各字段释义如下。

字段释义
clientid客户端 ID
disconnected_at终端连接断开时间 (s)
event事件类型,固定为 "client.disconnected"
node事件触发所在节点
expiry_intervalMQTT Session 过期时间
socknameemqx 监听的 IPAddress 和 Port
timestamp事件触发时间 (ms)
username用户名
peername终端的 IPAddress 和 Port
reason客户端连接断开原因

常见客户端离线原因如下。

Reason 类型原因
normalMQTT 客户端正常断开
discarded因为相同 ClientID 的客户端上线且设置 clean_start = true
takeovered因为相同 ClientID 的客户端上线且设置 clean_start = false
normalMQTT 客户端正常断开
kicked服务端踢出,通过 REST API
keepalive_timeoutMQTT keepalive 超时
not_authorized认证失败,或者 acl_nomatch = disconnect 时没有权限的 Pub/Sub 会主动断开客户端
tcp_closed客户端主动关闭 TCP 连接
receive_maximum_exceeded客户端发 qos2 消息,由于没有及时回复 PUBREL,导致 Broker 端堆积大量 PUBREC 报文,当堆积报文数达到上限,EMQX 将会主动断开连接
internal_error畸形报文解析出错
einvalEMQX 想向客户端发送一条消息,但是 Socket 已经断开
function_clauseMQTT 报文格式错误
etimedoutTCP 发送超时(没有收到 TCP ACK 回应)
proto_unexpected_c在已经有一条 MQTT 连接的情况下重复收到了 MQTT 连接请求
idle_timeoutTCP 连接建立 15s 之后,还没收到 connect 报文
其他可参考 MQTT 原因码与 ACK

设备消息通知

消息投递

消息投递的消息主题为 $events/message_delivered。在规则创建页面中输入如下规则匹配 SQL 语句便可完成消息投递规则的创建,当有新消息被投递便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/message_delivered"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 消息投递

各字段释义如下。

字段释义
clientid客户端 ID
event事件类型,固定为 "message.delivered"
flagsMQTT 消息的 Flags
from_clientid消息来源客户端 ID
from_username消息来源用户名
idMQTT 消息 ID
node事件触发所在节点
payloadMQTT 消息体
peerhost客户端的 IPAddress
publish_received_atPUBLISH 消息到达 Broker 的时间 (ms)
qosMQTT 消息的 QoS
timestamp事件触发时间 (ms)
topicMQTT 主题
username用户名

消息确认

消息确认的消息主题为 $events/message_acked。在规则创建页面中输入如下规则匹配 SQL 语句便可完成消息确认规则的创建,当有新消息被确认便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/message_acked"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 消息确认

各字段释义如下。

字段释义
clientid客户端 ID
event事件类型,固定为 "message.acked"
flagsMQTT 消息的 Flags
from_clientid消息来源客户端 ID
from_username消息来源用户名
idMQTT 消息 ID
node事件触发所在节点
payloadMQTT 消息体
peerhost客户端的 IPAddress
publish_received_atPUBLISH 消息到达 Broker 的时间 (ms)
qosMQTT 消息的 QoS
timestamp事件触发时间 (ms)
topicMQTT 主题
username用户名

消息在转发的过程中被丢弃

消息丢弃的消息主题为 $events/message_dropped。在规则创建页面中输入如下规则匹配 SQL 语句便可完成消息丢弃规则的创建,当有消息被丢弃便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/message_dropped"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 消息丢弃

各字段释义如下。

字段释义
clientid客户端 ID
reason消息丢弃原因,可能的原因:
no_subscribers: 没有订阅者
receive_maximum_exceeded: awaiting_rel 队列已满
packet_identifier_inuse: 消息 ID 已被使用
clientid消息来源 Client ID
username消息来源用户名
payloadMQTT 消息体
peerhost客户端的 IPAddress
topicMQTT 主题
qosMQTT 消息的 QoS
flagsMQTT 消息的 Flags
pub_propsPUBLISH Properties (仅适用于 MQTT 5.0)
timestamp事件触发时间 (ms)
publish_received_atPUBLISH 消息到达 Broker 的时间 (ms)
node事件触发所在节点

消息在投递的过程中被丢弃

消息丢弃的消息主题为 $events/delivery_dropped。在规则创建页面中输入如下规则匹配 SQL 语句便可完成消息丢弃规则的创建,当订阅者的消息队列满导致消息被丢弃便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/delivery_dropped"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 消息丢弃

各字段释义如下。

字段解释
idMQTT 消息 ID
reason消息丢弃原因,可能的原因:
queue_full: 消息队列已满(QoS>0)
no_local: 不允许客户端接收自己发布的消息
expired: 消息或者会话过期
qos0_msg: QoS0 的消息因为消息队列已满被丢弃
from_clientid消息来源 Client ID
from_username消息来源用户名
clientid消息目的 Client ID
username消息目的用户名
payloadMQTT 消息体
peerhost客户端的 IPAddress
topicMQTT 主题
qosMQTT 消息的 QoS
flagsMQTT 消息的 Flags
pub_propsPUBLISH Properties (仅适用于 MQTT 5.0)
timestamp事件触发时间 (ms)
publish_received_atPUBLISH 消息到达 Broker 的时间 (ms)
node事件触发所在节点

设备主题订阅通知

设备订阅

设备订阅消息的主题为 $events/session_subscribed。在规则创建页面中输入如下规则匹配 SQL 语句便可完成设备订阅规则的创建,当设备订阅主题便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/session_subscribed"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 设备订阅

各字段释义如下。

字段释义
clientid客户端 ID
event事件类型,固定为 "session.subscribed"
node事件触发所在节点
peerhost客户端的 IPAddress
qosMQTT 消息的 QoS
timestamp事件触发时间 (ms)
topicMQTT 主题
username用户名

设备取消订阅

设备取消订阅消息的主题为 $events/session_unsubscribed。在规则创建页面中输入如下规则匹配 SQL 语句便可完成设备取消订阅规则的创建,当设备取消订阅主题便会触发该规则。

sql
 SELECT
     *
 FROM
     "$events/session_unsubscribed"

使用 SQL 测试,查看触发该规则后返回字段,也可在 SQL 语句中使用 SELECT 和 WHERE 子句进行筛选与过滤。 设备取消订阅

各字段释义如下。

字段释义
clientid客户端 ID
event事件类型,固定为 "session.unsubscribed"
node事件触发所在节点
peerhost客户端的 IPAddress
qosMQTT 消息的 QoS
timestamp事件触发时间 (ms)
topicMQTT 主题
username用户名