使用emqx中mqtt服务的消息

nobita · 2026 年7 月 1 日 09:52

创建了一个mqtt服务链接，创建规则将接收到的属性上报消息转发到该连接器对应的主题，以及创建规则将该连接器对应主题的指令写入自己的主题。当长时间运行后，指令下发的规则未生效了，获取不到输入，属性上报的没问题。停用规则和连接器在启用后又可以了。。

zhongwencool · 2026 年7 月 1 日 10:15

用的什么版本，有相关的emqx日志么

nobita · 2026 年7 月 1 日 11:12

用的emqx 6.0.0,创建的规则的日志追踪中没有日志记录。容器控制台的日志都是如下:
2026-07-01T09:17:07.556890+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:46702, reason: {shutdown,tcp_closed}
2026-07-01T09:17:07.557121+00:00 [info] msg: terminate, peername: 192.168.200.132:46702, reason: {shutdown,tcp_closed}
2026-07-01T09:17:07.557112+00:00 [info] msg: terminate, peername: 192.168.200.132:46700, reason: {shutdown,tcp_closed}
2026-07-01T09:17:12.558069+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:56452, reason: {shutdown,tcp_closed}
2026-07-01T09:17:12.558069+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:56450, reason: {shutdown,tcp_closed}
2026-07-01T09:17:12.558330+00:00 [info] msg: terminate, peername: 192.168.200.132:56450, reason: {shutdown,tcp_closed}
2026-07-01T09:17:12.558330+00:00 [info] msg: terminate, peername: 192.168.200.132:56452, reason: {shutdown,tcp_closed}
2026-07-01T09:17:13.057051+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.27194>”
2026-07-01T09:17:13.057311+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.27195>” peer=“172.16.37.249:60196” action=ignore
2026-07-01T09:17:14.932101+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.1743>”
2026-07-01T09:17:14.932160+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.1744>” peer=“172.16.37.249:43972” action=ignore
2026-07-01T09:17:17.445070+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.27032>”
2026-07-01T09:17:17.445107+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.27033>” peer=“172.16.37.249:52168” action=ignore
2026-07-01T09:17:17.556576+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:56466, reason: {shutdown,tcp_closed}
2026-07-01T09:17:17.556576+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:56482, reason: {shutdown,tcp_closed}
2026-07-01T09:17:17.556773+00:00 [info] msg: terminate, peername: 192.168.200.132:56466, reason: {shutdown,tcp_closed}
2026-07-01T09:17:17.556830+00:00 [info] msg: terminate, peername: 192.168.200.132:56482, reason: {shutdown,tcp_closed}
2026-07-01T09:17:18.601057+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.1745>”
2026-07-01T09:17:18.601070+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.1746>” peer=“172.16.37.249:43976” action=ignore
2026-07-01T09:17:22.556626+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:42648, reason: {shutdown,tcp_closed}
2026-07-01T09:17:22.556832+00:00 [info] msg: terminate, peername: 192.168.200.132:42648, reason: {shutdown,tcp_closed}
2026-07-01T09:17:22.556721+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:42636, reason: {shutdown,tcp_closed}
2026-07-01T09:17:22.556944+00:00 [info] msg: terminate, peername: 192.168.200.132:42636, reason: {shutdown,tcp_closed}
2026-07-01T09:17:27.556487+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:42654, reason: {shutdown,tcp_closed}
2026-07-01T09:17:27.557128+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:42656, reason: {shutdown,tcp_closed}
2026-07-01T09:17:27.557166+00:00 [info] msg: terminate, peername: 192.168.200.132:42654, reason: {shutdown,tcp_closed}
2026-07-01T09:17:27.557393+00:00 [info] msg: terminate, peername: 192.168.200.132:42656, reason: {shutdown,tcp_closed}
2026-07-01T09:17:29.399124+00:00 [debug] msg: send_data, packet: PINGREQ(Q0, R0, D0), socket: #Port<0.1268275>, payload_encode: text
2026-07-01T09:17:29.433472+00:00 [debug] msg: recv_data, data: Encoded(text)=<<208,0>>, payload_encode: text
2026-07-01T09:17:32.556370+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:40478, reason: {shutdown,tcp_closed}
2026-07-01T09:17:32.556400+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:40494, reason: {shutdown,tcp_closed}
2026-07-01T09:17:32.556606+00:00 [info] msg: terminate, peername: 192.168.200.132:40494, reason: {shutdown,tcp_closed}
2026-07-01T09:17:32.556594+00:00 [info] msg: terminate, peername: 192.168.200.132:40478, reason: {shutdown,tcp_closed}
2026-07-01T09:17:35.624831+00:00 [debug] msg: auth_cache_cleanup, name: emqx_authz_cache, num_deleted: 0
2026-07-01T09:17:35.847974+00:00 [debug] msg: auth_cache_cleanup, name: emqx_authn_cache, num_deleted: 0
2026-07-01T09:17:36.267171+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.894>”
2026-07-01T09:17:36.267271+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.895>” peer=“172.16.37.249:51012” action=ignore
2026-07-01T09:17:37.556463+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:40496, reason: {shutdown,tcp_closed}
2026-07-01T09:17:37.556464+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:40498, reason: {shutdown,tcp_closed}
2026-07-01T09:17:37.556667+00:00 [info] msg: terminate, peername: 192.168.200.132:40496, reason: {shutdown,tcp_closed}
2026-07-01T09:17:37.556675+00:00 [info] msg: terminate, peername: 192.168.200.132:40498, reason: {shutdown,tcp_closed}
2026-07-01T09:17:38.785043+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.30486>”
2026-07-01T09:17:38.785129+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.30487>” peer=“172.16.37.249:50556” action=ignore
2026-07-01T09:17:39.135033+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.616>”
2026-07-01T09:17:39.135977+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.617>” peer=“172.16.37.249:44218” action=ignore
2026-07-01T09:17:42.556641+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:43722, reason: {shutdown,tcp_closed}
2026-07-01T09:17:42.556798+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:43736, reason: {shutdown,tcp_closed}
2026-07-01T09:17:42.556906+00:00 [info] msg: terminate, peername: 192.168.200.132:43722, reason: {shutdown,tcp_closed}
2026-07-01T09:17:42.556940+00:00 [info] msg: terminate, peername: 192.168.200.132:43736, reason: {shutdown,tcp_closed}
2026-07-01T09:17:46.882111+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.1016>”
2026-07-01T09:17:46.882552+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.1017>” peer=“172.16.37.249:32882” action=ignore
2026-07-01T09:17:47.556921+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:43760, reason: {shutdown,tcp_closed}
2026-07-01T09:17:47.556921+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:43752, reason: {shutdown,tcp_closed}
2026-07-01T09:17:47.557135+00:00 [info] msg: terminate, peername: 192.168.200.132:43752, reason: {shutdown,tcp_closed}
2026-07-01T09:17:47.557132+00:00 [info] msg: terminate, peername: 192.168.200.132:43760, reason: {shutdown,tcp_closed}
2026-07-01T09:17:48.232072+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.805>”
2026-07-01T09:17:48.232442+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.806>” peer=“172.16.37.249:39306” action=ignore
2026-07-01T09:17:50.733101+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.813>”
2026-07-01T09:17:50.733102+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.814>” peer=“172.16.37.249:39314” action=ignore
2026-07-01T09:17:51.919462+00:00 [debug] clientid: KT4G261C0174, msg: cancel_stats_timer, peername: 172.16.219.64:62667, username: crucgdev
2026-07-01T09:17:51.996098+00:00 [debug] tag: MQTT, clientid: KT4G261C0174, msg: raw_bin_received, peername: 172.16.219.64:62667, username: crucgdev, size: 2, type: hex, bin: C000
2026-07-01T09:17:51.996406+00:00 [debug] tag: MQTT, clientid: KT4G261C0174, msg: mqtt_packet_received, peername: 172.16.219.64:62667, username: crucgdev, packet: PINGREQ(Q0, R0, D0), payload_encode: text
2026-07-01T09:17:51.996570+00:00 [debug] tag: MQTT, clientid: KT4G261C0174, msg: mqtt_packet_sent, peername: 172.16.219.64:62667, username: crucgdev, packet: PINGRESP(Q0, R0, D0), payload_encode: text
2026-07-01T09:17:52.556769+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:59952, reason: {shutdown,tcp_closed}
2026-07-01T09:17:52.556968+00:00 [debug] tag: SOCKET, msg: emqx_connection_terminated, peername: 192.168.200.132:59950, reason: {shutdown,tcp_closed}
2026-07-01T09:17:52.557100+00:00 [info] msg: terminate, peername: 192.168.200.132:59952, reason: {shutdown,tcp_closed}
2026-07-01T09:17:52.557130+00:00 [info] msg: terminate, peername: 192.168.200.132:59950, reason: {shutdown,tcp_closed}

zhongwencool · 2026 年7 月 1 日 14:03

先升到 6.0.2
你现在这几条线索是同一个方向

规则日志追踪没有记录：消息没有进入这条规则，不是 SQL 或动作执行失败。
日志里反复出现 tcp_closed：MQTT 服务连接发生过断开/重连。
停用规则和连接器再启用后恢复：等价于重新建立 Source 订阅。
6.0.2 变更里对应的是 #16507：修复 MQTT Source 在 Connector 重新连接后停止接收消息。之前 Connector 从连接丢失恢复后，Source 的 topic 没有重新订阅，就会变成“连接看起来恢复了，但 Source 不再收到远端消息”，直到重启 Source/连接器。
先做两步：

升级到 6.0.2 或更高版本复测。
如果短期不能升级，临时规避就是发现指令下发无输入后重启 MQTT Source/连接器；这只能绕过
升级后这样验证：手动断开远端 MQTT 服务再恢复，然后往指令主题发一条消息，看规则追踪里是否出现 Source 输入。

nobita · 2026 年7 月 1 日 14:55

好的，这边试下感谢