EMQX消息流入速率和流出速率异常

[shenjinyong]

环境

• EMQX 版本：
• 操作系统版本：
  Ubuntu 22.04

重现此问题的步骤

1. emqx崩溃重启
2. xxx
3. xxx

预期行为

消息流入速率和流出速率与实际一致

实际行为

消息流入速率机会没用
消息流出速率疯狂增大，100、 120、 到1000

2025-09-22T17:21:43.993786+08:00 [error] msg: connection_error, mfa: emqtt:handle_event/4(1416), error: tcp_error, reason: etimedout
2025-09-22T17:21:50.947129+08:00 [warning] msg: alarm_is_activated, mfa: emqx_alarm:do_actions/3(418), message: <<“resource down: Unknown reason”>>, name: <<“connector:mqtt:mqtt”>>
2025-09-22T17:21:50.947600+08:00 [warning] msg: health_check_failed, mfa: emqx_resource_manager:handle_connected_health_check/1(893), id: <<“connector:mqtt:mqtt”>>, status: disconnected
2025-09-22T17:21:50.947878+08:00 [warning] msg: alarm_is_activated, mfa: emqx_alarm:do_actions/3(418), message: <<“resource down: #{error => <<"Resource connector:mqtt:mqtt for channel action:mqtt:iot_rule_line:connector:mqtt:mqtt is not connected. Resource status: disconnected">>,status => disconnected}”>>, name: <<“action:mqtt:iot_rule_line:connector:mqtt:mqtt”>>
2025-09-22T17:21:50.948168+08:00 [warning] msg: alarm_is_activated, mfa: emqx_alarm:do_actions/3(418), message: <<“resource down: #{error => <<"Resource connector:mqtt:mqtt for channel action:mqtt:iot_rule_offline:connector:mqtt:mqtt is not connected. Resource status: disconnected">>,status => disconnected}”>>, name: <<“action:mqtt:iot_rule_offline:connector:mqtt:mqtt”>>
2025-09-22T17:21:50.948420+08:00 [warning] msg: alarm_is_activated, mfa: emqx_alarm:do_actions/3(418), message: <<“resource down: #{error => <<"Resource connector:mqtt:mqtt for channel action:mqtt:iot_rule_online:connector:mqtt:mqtt is not connected. Resource status: disconnected">>,status => disconnected}”>>, name: <<“action:mqtt:iot_rule_online:connector:mqtt:mqtt”>>
2025-09-22T17:22:05.984832+08:00 [warning] msg: alarm_is_deactivated, mfa: emqx_alarm:do_actions/3(424), name: <<“action:mqtt:iot_rule_line:connector:mqtt:mqtt”>>
2025-09-22T17:22:05.985244+08:00 [warning] msg: alarm_is_deactivated, mfa: emqx_alarm:do_actions/3(424), name: <<“action:mqtt:iot_rule_offline:connector:mqtt:mqtt”>>
2025-09-22T17:22:05.985492+08:00 [warning] msg: alarm_is_deactivated, mfa: emqx_alarm:do_actions/3(424), name: <<“action:mqtt:iot_rule_online:connector:mqtt:mqtt”>>
2025-09-22T17:22:05.985757+08:00 [warning] msg: alarm_is_deactivated, mfa: emqx_alarm:do_actions/3(424), name: <<“connector:mqtt:mqtt”>>

[zhongwencool]

@shenjinyong

消息流入速率机会没用

什么叫机会没用

@shenjinyong

速率疯狂增大，100、 120、 到1000

1对n就会这样扇型流速，（发布一条n个端订阅）。不算异常吧。
日志中的报错原因是mqtt桥接对端连接失败了

[shenjinyong]

1、EMQX 流入数量几乎没有 但是流出有几百个每秒
2、实际流入数量不止每秒不到1个
3、实际流出页没有你们多

[shenjinyong]

[zhongwencool]

应该不会吧，
如果是这个么情况，那应该是致命bug…
麻烦提供一下可复现的步骤（代码），我们安排复现一下。

[shenjinyong]

目前生产环境统计情况就是这样的，实际上我们也没有进行什么操作，不知道发生的原因是什么，只是从前段时间emqx自动重启后就出现这样的情况

[zhongwencool]

也就是说重启前是正常的？
无法重现，这就很难定位了

[shenjinyong]

应该说之前是正常的 然后出现异常情况导致EMQX崩溃重启，然后就异常了。而且现在实际应该还是正常的流入流出，只是EMQX上统计的数据不对。同时还有异常情况，导致规则的在线离线不正常通知了，查看依旧有比较多的异常告警。

[shenjinyong]

目前又出现诡异的现象了，消息流出突然呈现线性增长

[zhongwencool]

这张图里的两组数据能对上：EMQX 5.5.0 底部历史图默认记录每 10 秒的消息增量，末端约 17,000 条/10 秒，对应上面的 1,726 条/秒；流入约 50 条/10 秒，也对应 5 条/秒。

先看看这三类原因：

• 发布主题命中了大量订阅，或者存在重叠的 #、+ 通配符订阅。
• MQTT 桥接或规则的 republish 输出又被订阅、桥接回来了，形成消息回环。
• QoS 1/2 消费端没有正常回复 ACK，导致 PUBLISH 持续重发。

截图里消息丢弃也在周期性尖峰，需要同时确认具体是哪项指标增长。连续执行两次，间隔 30 秒，把相关指标及差值贴出来：

./bin/emqx ctl broker metrics | grep -E 'messages\.(received|sent|qos[012]\.(received|sent)|acked|dropped)|delivery\.dropped|packets\.(publish\.sent|puback\.received|pubrec\.received|pubcomp\.received)'

[shenjinyong]

1、这个是我们真实的环境，因此没有命中大量订阅的情况
2、这个发现是每天的这个时间段会这样

[zhongwencool]

那就奇怪了。那能按上面的命令执行看看么

[shenjinyong]

您，已经截图了并统计相关指标差值

[zhongwencool]

通过这些差值可以明确了：EMQX 在大量发送 QoS 1 消息，而且绝大多数没有及时收到 PUBACK。

按前面约 30 秒的取样：

• messages.received 增加 268，messages.sent 增加 68,054，出入比约 254 倍。
• messages.qos1.sent 增加 67,873，占流出消息的 99.7%。
• packets.puback.received 和 messages.acked 都只增加 174。
• messages.dropped、delivery.dropped 没有增长，和丢弃无关。

EMQX 5.5.0 的 retry_interval 默认是 30 秒。这组数据很像部分订阅客户端没有回复 PUBACK，飞行窗口里的 QoS 1 消息被周期性重传；每天固定时段出现，优先查该时段在线的订阅端或网络，不是发布端流量。

我看你的连接数并不算很多，可以在在异常时段、同一节点执行：

/usr/bin/emqx ctl clients list \
  | grep -E 'inflight=[1-9][0-9]*|enqueued_msgs=[1-9][0-9]*' \
  | head -100

把输出贴出来，重点看 clientid、peername、inflight、enqueued_msgs 和 subscriptions。先定位是哪批客户端不回 ACK，再针对这些客户端查网络和消费逻辑。

[shenjinyong]

@zhongwencool

数并不算很多，可以在在异常时段、同一节点执行：

我们当时查看 ，每一个都是这样

Client(, username=admin, peername=, clean_start=false, keepalive=60, session_expiry_interval=43200, subscriptions=6, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784095131646, connected_at=1784754910089) 每一个都是这样

[shenjinyong]

当时的截图可以看到每一个enqueued_msgs都是0

[shenjinyong]

这个是指令的执行结果

[zhongwencool]

命令没有输出，说明执行时所有客户端的 inflight 和 enqueued_msgs 都是 0。EMQX 的重传消息一定还在 inflight 窗口里，所以前面“不回 PUBACK 导致持续重传”的判断应该是错的，sorry…

下一次消息流出曲线开始上升时，在这台机器连续抓两份同步快照：

out="/tmp/emqx-flow-$(date +%Y%m%d-%H%M%S)"
mkdir -p "$out"

/usr/bin/emqx ctl status > "$out/status.txt"
/usr/bin/emqx ctl subscriptions list > "$out/subscriptions.txt"

for i in 1 2; do
  date -Ins > "$out/time.$i.txt"
  /usr/bin/emqx ctl broker metrics > "$out/metrics.$i.txt"
  /usr/bin/emqx ctl clients list > "$out/clients.$i.txt"
  [ "$i" = 1 ] && sleep 10
done

diff -u "$out/metrics.1.txt" "$out/metrics.2.txt" > "$out/metrics.diff.txt" || true
diff -u "$out/clients.1.txt" "$out/clients.2.txt" > "$out/clients.diff.txt" || true

tar -czf "$out.tgz" -C "$(dirname "$out")" "$(basename "$out")"
echo "$out.tgz"

把压缩包里的 metrics.diff.txt、clients.diff.txt 和 subscriptions.txt 脱敏后贴出来。

• messages.sent、messages.qos1.sent 和客户端 delivered_msgs 同时增长：是真实投递，再根据变化的 client 和订阅查扇出。
• messages.sent 增长、messages.delivered 不增长，同时出现非零 inflight：才是 QoS 1 重传。
• messages.sent 增长，但 messages.delivered、所有客户端的 delivered_msgs、inflight 都不增长：更像 5.5.0 的指标统计异常，需要保留这组证据后升级验证。

另外对比 messages.publish：如果它也在固定时段增长，优先查业务定时任务，以及现有 MQTT Connector 和在线/离线规则是否形成回环。现在无法根据单张 Dashboard 图猜，先把发送、投递和客户端状态放在同一个 10 秒窗口里对上。

[shenjinyong]

【metrics.diff.txt】

— /tmp/emqx-flow-20260724-103653/metrics.1.txt 2026-07-24 10:36:57.821319952 +0800
+++ /tmp/emqx-flow-20260724-103653/metrics.2.txt 2026-07-24 10:37:10.626337392 +0800
@@ -1,23 +1,23 @@
authentication.failure : 0
-authentication.success : 591232
-authentication.success.anonymo: 591232
-authorization.allow : 3688828
-authorization.cache_hit : 2526350
-authorization.cache_miss : 1162532
+authentication.success : 591250
+authentication.success.anonymo: 591250
+authorization.allow : 3688884
+authorization.cache_hit : 2526375
+authorization.cache_miss : 1162563
authorization.deny : 54
-authorization.matched.allow : 1162473
+authorization.matched.allow : 1162504
authorization.matched.deny : 54
authorization.nomatch : 0
authorization.superuser : 0
-bytes.received : 63229645448
-bytes.sent : 71206518380
+bytes.received : 63230839243
+bytes.sent : 71213240253
client.auth.anonymous : 0
-client.authenticate : 591232
-client.authorize : 1162532
-client.connack : 591232
-client.connect : 591232
-client.connected : 591232
-client.disconnected : 591314
+client.authenticate : 591250
+client.authorize : 1162563
+client.connack : 591250
+client.connect : 591250
+client.connected : 591250
+client.disconnected : 591332
client.subscribe : 7891
client.unsubscribe : 6
delivery.dropped : 3
@@ -26,22 +26,22 @@
delivery.dropped.qos0_msg : 3
delivery.dropped.queue_full : 0
delivery.dropped.too_large : 0
-messages.acked : 2001675
+messages.acked : 2001677
messages.delayed : 0
-messages.delivered : 59509068
+messages.delivered : 59524610
messages.dropped : 609528
messages.dropped.await_pubrel_: 0
messages.dropped.no_subscriber: 609528
messages.forward : 0
-messages.publish : 3680452
-messages.qos0.received : 3573363
-messages.qos0.sent : 1921974
-messages.qos1.received : 107089
-messages.qos1.sent : 57587094
-messages.qos2.received : 529
+messages.publish : 3680500
+messages.qos0.received : 3573410
+messages.qos0.sent : 1922021
+messages.qos1.received : 107090
+messages.qos1.sent : 57602589
+messages.qos2.received : 537
messages.qos2.sent : 0
-messages.received : 3680981
-messages.sent : 59509068
+messages.received : 3681037
+messages.sent : 59524610
overload_protection.delay.ok : 0
overload_protection.delay.time: 0
overload_protection.gc : 0
@@ -51,16 +51,16 @@
packets.auth.sent : 0
packets.connack.auth_error : 0
packets.connack.error : 0
-packets.connack.sent : 591232
-packets.connect.received : 591232
+packets.connack.sent : 591250
+packets.connect.received : 591250
packets.disconnect.received : 13
-packets.disconnect.sent : 520
-packets.pingreq.received : 1377825
-packets.pingresp.sent : 1377825
+packets.disconnect.sent : 528
+packets.pingreq.received : 1377846
+packets.pingresp.sent : 1377846
packets.puback.inuse : 0
packets.puback.missed : 8395
-packets.puback.received : 2010070
-packets.puback.sent : 107089
+packets.puback.received : 2010072
+packets.puback.sent : 107090
packets.pubcomp.inuse : 0
packets.pubcomp.missed : 0
packets.pubcomp.received : 0
@@ -69,8 +69,8 @@
packets.publish.dropped : 0
packets.publish.error : 0
packets.publish.inuse : 0
-packets.publish.received : 3680981
-packets.publish.sent : 59509068
+packets.publish.received : 3681037
+packets.publish.sent : 59524610
packets.pubrec.inuse : 0
packets.pubrec.missed : 0
packets.pubrec.received : 0
@@ -78,8 +78,8 @@
packets.pubrel.missed : 0
packets.pubrel.received : 0
packets.pubrel.sent : 0
-packets.received : 7668018
-packets.sent : 61593631
+packets.received : 7668115
+packets.sent : 61609221
packets.suback.sent : 7891
packets.subscribe.auth_error : 0
packets.subscribe.error : 0
@@ -87,8 +87,8 @@
packets.unsuback.sent : 6
packets.unsubscribe.error : 0
packets.unsubscribe.received : 6
-session.created : 590504
-session.discarded : 586088
+session.created : 590522
+session.discarded : 586098
session.resumed : 728
session.takenover : 728
-session.terminated : 4271
+session.terminated : 4279

【clients.diff】

— /tmp/emqx-flow-20260724-103653/clients.1.txt 2026-07-24 10:36:59.226319647 +0800
+++ /tmp/emqx-flow-20260724-103653/clients.2.txt 2026-07-24 10:37:12.068337798 +0800
@@ -129,7 +129,7 @@
Client(, username=, peername=, clean_start=false, keepalive=60, session_expiry_interval=43200, subscriptions=1, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784130444865, connected_at=1784216080881)
Client(, username=, peername=, clean_start=false, keepalive=60, session_expiry_interval=43200, subscriptions=1, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784751707896, connected_at=1784751707896)
Client(.1_202602030004, username=, peername=, clean_start=false, keepalive=60, session_expiry_interval=43200, subscriptions=6, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784095132147, connected_at=1784095183478)
-Client(, username=, peername=, clean_start=true, keepalive=20, session_expiry_interval=0, subscriptions=0, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784860619183, connected_at=1784860619183)
+Client(, username=, peername=, clean_start=true, keepalive=20, session_expiry_interval=0, subscriptions=0, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784860630935, connected_at=1784860630935)
Client(, username=, peername=, clean_start=false, keepalive=60, session_expiry_interval=43200, subscriptions=6, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784095132248, connected_at=1784423984151)
Client(, username=, peername=, clean_start=true, keepalive=60, session_expiry_interval=0, subscriptions=4, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784817091510, connected_at=1784817091510)
Client(, username=, peername=, clean_start=false, keepalive=60, session_expiry_interval=43200, subscriptions=6, inflight=0, awaiting_rel=0, delivered_msgs=0, enqueued_msgs=0, dropped_msgs=0, connected=true, created_at=1784095132056, connected_at=1784095186398)

【subscriptions.txt】

[shenjinyong]

另外我们查看单个客户端没有统计到消息流入和流出的数据异常，要么都是0要么只有一个qos0流入值为1，很多个客户端都这样，实际上设备是在上报数据的