设备mqtt经常重连+消费消息失败

s264 · 2026 年2 月 26 日 09:53

环境

EMQX 版本：开源版emqx:5.8.7
操作系统版本：centos:7

这是某个设备的日志信息，目前的情况是1、设备经常重连2、消费qos2消息失败，我不知道是消费消息的时候设备重连的问题导致再次重连会话出问题还是怎么情况，我只看出来packet_id=32996有问题。设备端代码是cleanSession设置成false 然后消费的消息级别是qos2，帮忙减排查下问题谢谢。

zhongwencool · 2026 年2 月 26 日 11:04

优先排查同 clientId 重复连接导致 session takeover，QoS2 事务被打断后 packet_id 复用。
先做这几件事：

确认是否有多个设备/线程用同一个 clientId（EMQX 日志里会有 takenover/被踢）。
把截图里的日志原文贴出来（包含断开原因/错误码），图片太小看不清。
贴 EMQX 侧该 clientId 的日志片段。
可直接执行：
grep -n “clientid=你的ClientId” /path/to/emqx.log | tail -n 200
emqx ctl clients lookup 你的ClientId
还需要这些信息：
使用的客户端 SDK/版本、MQTT 版本（3.1.1/5）
keepalive、cleanSession/clean_start、session_expiry_interval
是否有负载均衡/网关（会不会把同一 clientId 分到不同后端）
如果确实频繁断线，QoS2 会卡在 PUBREC/PUBREL 流程里，重连时 packet_id 复用就会报错。EMQX 这边也可以看下 inflight 配置是否过小：
mqtt.max_inflight
mqtt.max_awaiting_rel

s264 · 2026 年2 月 27 日 01:56

确认没有多个设备/线程用同一个 clientId 因为我们的设备都是用sn作为clientId使用然后嵌入式代码都是用一个线程重连
这是日志原文也是通过面板工具的日志跟踪该clientId的
emqx@12-trace_A0126010127_2026-02-10.zip (2.0 KB)
3.客户端 SDK/版本嵌入式使用的库是直接使用https://github.com/eclipse-paho/paho.mqtt.embedded-c 改地址直接下载下来使用的 MQTT 版本是3.1.1
4.没有负载均衡/网关因为现在使用的单nginx、单点emqx服务不存在把同一 clientId 分到不同后端
5.如果确实频繁断线，QoS2 会卡在 PUBREC/PUBREL 流程里这个问题因为我们设备是连接的WiFi网络而且是在办公室连接的信号都是满格测试的，这个是否可以通过修改嵌入式代码解决频繁断线我可以提供代码
6、mqtt.max_inflight
mqtt.max_awaiting_rel 这2个都是默认的32 和100 我们测试的时候只发送了几条消息测试所以消息数量可以确认没有达到这个数量级
这是嵌入式的参数配置信息 MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer;
connectData.MQTTVersion = 4;
connectData.clientID.cstring = MQTT_CLIENTID;
connectData.cleansession = 0;
connectData.keepAliveInterval = 60;
client.defaultMessageHandler = messageArrived;
至于 session_expiry_interval没有在代码里面找到该参数

zhongwencool · 2026 年2 月 27 日 02:41

你这类 QoS2 失败基本是“断线导致 QoS2 四步被打断 + session 恢复后 packet_id 复用”，不是 EMQX 配置问题。MQTT 3.1.1 没有 session_expiry_interval，cleanSession=0 就是持久会话。
先做这几步验证/修：

临时把 cleansession=1 跑一轮，看 QoS2 失败是否消失（用来确认是持久会话+断线导致的）。
确认你的 embedded-c 实现是否启用持久会话的 inflight 持久化。默认只是内存，断线重连后 packet_id 可能重用，必须存储未完成的 QoS2 状态。
确保 MQTTYield/MQTTClient_yield 频率足够（否则 keepalive 超时被踢），并把日志里的断开原因贴出来。
：

设备端完整的 QoS2 报错行（含 packet_id）
EMQX emqx.log 里该 clientId 的断开原因（reason/takenover/keepalive_timeout 等）
你们的重连逻辑代码片段（重连前是否清理/恢复 session 状态）
另外确认没有“同一 sn 多实例同时在线”（比如重启没等旧连接断开），这会直接触发 session takeover。

s264 · 2026 年2 月 28 日 03:20

1、临时把 cleansession=1 跑一轮 qos2能够正常消费
2、 embedded-c 没有启用持久会话的 inflight 持久化
3、 MQTTYield/MQTTClient_yield 设置的100ms
4、这是详细日志：2026-02-11T09:21:05.306849+00:00 [MQTT] testSN@118:23760 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=64, Payload(text)={“type”:1,“data”:{“quantityElectricity”:38,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:21:05.307091+00:00 [PUBLISH] testSN@118:23760 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:38,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:21:05.307228+00:00 [MQTT] testSN@118:23760 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=64, ReasonCode=0), username: testName
2026-02-11T09:21:53.376704+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0, ClientId=testSN, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=testName, Password=******), username: testName
2026-02-11T09:21:53.376904+00:00 [AUTHN] testSN@118:25063 msg: authenticator_result, authenticator: password_based:built_in_database, result: {ok,#{is_superuser => false}}, username: testName
2026-02-11T09:21:53.377014+00:00 [AUTHN] testSN@118:25063 msg: authentication_result, reason: chain_result, result: {stop,{ok,#{is_superuser => false}}}, username: testName
2026-02-11T09:21:53.377766+00:00 [SOCKET] testSN@118:23760 msg: emqx_connection_terminated, reason: {shutdown,takenover}, username: testName
2026-02-11T09:21:53.380152+00:00 [warning] clientid: testSN, msg: emqx_persistent_session_ds_replay_inconsistency, peername: 118:25063, username: testName, expected: {srs,<<“13”>>,13,#{1 => 61,2 => <<“13”>>,3 => #{1 => 2,2 => 13,3 => {it,<<102,82,247,27,74,40,30,78>>,1770603683293999,<<“testSN”>>,}}},#{1 => 61,2 => <<“13”>>,3 => #{1 => 2,2 => 13,3 => {it,<<102,82,247,27,74,40,30,78>>,1770796779383000,<<“testSN”>>,}}},1,18,227,18,228,false,3}, got: {srs,<<“13”>>,13,#{1 => 61,2 => <<“13”>>,3 => #{1 => 2,2 => 13,3 => {it,<<102,82,247,27,74,40,30,78>>,1770603683293999,<<“testSN”>>,}}},#{1 => 61,2 => <<“13”>>,3 => #{1 => 2,2 => 13,3 => {it,<<102,82,247,27,74,40,30,78>>,1770790304944000,<<“testSN”>>,}}},1,18,227,18,228,false,3}
2026-02-11T09:21:53.380946+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0, AckFlags=1, ReasonCode=0), username: testName
2026-02-11T09:21:53.381697+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0, PacketId=32996, ReasonCode=0), username: testName
2026-02-11T09:21:53.421357+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: SUBSCRIBE(Q1, R0, D0, PacketId=2 TopicFilters=[device/testSN(#{nl => 0,qos => 2,rap => 0,rh => 0})]), username: testName
2026-02-11T09:21:53.421568+00:00 [AUTHZ] testSN@118:25063 msg: authorization_module_ignore, action: SUBSCRIBE(Q2), authorize_type: client_info, module: emqx_authz_client_info, topic: device/testSN, username: testName
2026-02-11T09:21:53.421627+00:00 [AUTHZ] testSN@118:25063 msg: authorization_matched_allow, action: SUBSCRIBE(Q2), authorize_type: file, module: emqx_authz_file, topic: device/testSN, username: testName
2026-02-11T09:21:53.422323+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: SUBACK(Q0, R0, D0, PacketId=2, ReasonCodes=[2]), username: testName
2026-02-11T09:21:53.455666+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0, PacketId=32996, ReasonCode=0), username: testName
2026-02-11T09:21:53.455818+00:00 [warning] clientid: testSN, msg: out-of-order_commit, peername: 118:25063, username: testName, expected: undefined, seqno: 228, packet_id: 32996, track: pubcomp
2026-02-11T09:21:53.456193+00:00 [warning] clientid: testSN, msg: pubcomp_packetId_not_found, peername: 118:25063, username: testName, packetId: 32996
2026-02-11T09:21:56.558007+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=3, Payload(text)={“type”:4,“data”:{“type”:“D0004”,“sn”:“testSN”}}), username: testName
2026-02-11T09:21:56.558197+00:00 [AUTHZ] testSN@118:25063 msg: authorization_module_ignore, action: PUBLISH(Q1,R0), authorize_type: client_info, module: emqx_authz_client_info, topic: service/testSN, username: testName
2026-02-11T09:21:56.558258+00:00 [AUTHZ] testSN@118:25063 msg: authorization_matched_allow, action: PUBLISH(Q1,R0), authorize_type: file, module: emqx_authz_file, topic: service/testSN, username: testName
2026-02-11T09:21:56.558305+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:4,“data”:{“type”:“D0004”,“sn”:“testSN”}}
2026-02-11T09:21:56.558422+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=3, ReasonCode=0), username: testName
2026-02-11T09:21:59.404467+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=4, Payload(text)={“type”:4,“data”:{“type”:“D0004”,“sn”:“testSN”}}), username: testName
2026-02-11T09:21:59.404648+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:4,“data”:{“type”:“D0004”,“sn”:“testSN”}}
2026-02-11T09:21:59.404785+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=4, ReasonCode=0), username: testName
2026-02-11T09:22:08.052970+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=5, Payload(text)={“type”:1,“data”:{“quantityElectricity”:18,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:08.053273+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:18,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:08.053454+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=5, ReasonCode=0), username: testName
2026-02-11T09:22:14.314907+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=6, Payload(text)={“type”:1,“data”:{“quantityElectricity”:17,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:14.315240+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:17,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:14.315437+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=6, ReasonCode=0), username: testName
2026-02-11T09:22:20.355745+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=7, Payload(text)={“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:20.356060+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:20.356283+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=7, ReasonCode=0), username: testName
2026-02-11T09:22:26.393497+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=8, Payload(text)={“type”:1,“data”:{“quantityElectricity”:15,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:26.393766+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:15,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:26.393958+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=8, ReasonCode=0), username: testName
2026-02-11T09:22:32.433240+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=9, Payload(text)={“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:32.433477+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:32.433621+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=9, ReasonCode=0), username: testName
2026-02-11T09:22:38.676916+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=10, Payload(text)={“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:38.677122+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:38.677266+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=10, ReasonCode=0), username: testName
2026-02-11T09:22:44.814849+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=11, Payload(text)={“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:44.815155+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:44.815364+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=11, ReasonCode=0), username: testName
2026-02-11T09:22:50.874962+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=12, Payload(text)={“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:50.875195+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:50.875353+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=12, ReasonCode=0), username: testName
2026-02-11T09:22:56.926028+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=13, Payload(text)={“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:22:56.926352+00:00 [AUTHZ] testSN@118:25063 msg: authorization_module_ignore, action: PUBLISH(Q1,R0), authorize_type: client_info, module: emqx_authz_client_info, topic: service/testSN, username: testName
2026-02-11T09:22:56.926421+00:00 [AUTHZ] testSN@118:25063 msg: authorization_matched_allow, action: PUBLISH(Q1,R0), authorize_type: file, module: emqx_authz_file, topic: service/testSN, username: testName
2026-02-11T09:22:56.926495+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:22:56.926653+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=13, ReasonCode=0), username: testName
2026-02-11T09:23:02.990263+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=14, Payload(text)={“type”:1,“data”:{“quantityElectricity”:15,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:23:02.990524+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:15,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:23:02.990695+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=14, ReasonCode=0), username: testName
2026-02-11T09:23:09.076525+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=15, Payload(text)={“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:23:09.076744+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:23:09.076918+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=15, ReasonCode=0), username: testName
2026-02-11T09:23:15.119515+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=16, Payload(text)={“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:23:15.119751+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:23:15.119901+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=16, ReasonCode=0), username: testName
2026-02-11T09:23:21.354400+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=17, Payload(text)={“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:23:21.354699+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:23:21.354851+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=17, ReasonCode=0), username: testName
2026-02-11T09:23:27.398394+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=18, Payload(text)={“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:23:27.398637+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:14,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:23:27.398810+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=18, ReasonCode=0), username: testName
2026-02-11T09:23:33.466927+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=19, Payload(text)={“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:23:33.467187+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:23:33.467354+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=19, ReasonCode=0), username: testName
2026-02-11T09:23:39.508380+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_received, packet: PUBLISH(Q1, R0, D0, Topic=service/testSN, PacketId=20, Payload(text)={“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}), username: testName
2026-02-11T09:23:39.508612+00:00 [PUBLISH] testSN@118:25063 msg: publish_to, payload_encode: text, topic: service/testSN, username: testName, payload: {“type”:1,“data”:{“quantityElectricity”:13,“powerStatus”:2,“wifiSignalStrength”:3,“wifiName”:“ChinaNet”,“sn”:“testSN”,“volume”:6,“switchStatus”:2}}
2026-02-11T09:23:39.508763+00:00 [MQTT] testSN@118:25063 msg: mqtt_packet_sent, packet: PUBACK(Q0, R0, D0, PacketId=20, ReasonCode=0), username: testName
5、这是重连代码：void paho_mqtt_main(void)
{
mqtt_msg_id_t msg_id;
uint32_t last_try = 0;

MQTTDisconnect(&client);
BSD_disconnect(&network);

mqtt_send_msg_queue_init();

paho_mqtt_start();

while (1)
{
    /* ---------- 重连节流 ---------- */
    if (!client.isconnected && !mqtt_connecting)
    {
        if ((rtos_get_time() - last_try) > 5000)
        {
            mqtt_connecting = true;
            last_try = rtos_get_time();

            int rc = paho_mqtt_start();

            mqtt_connecting = false;

            if (rc != 0)
            {
                printf("MQTT connect failed rc=%d\n", rc);
            }
        }
    }

    /* ---------- MQTT 网络处理 ---------- */
    if (client.isconnected)
    {
        MQTTYield(&client, 100);

        if (!rtos_is_queue_empty(&mqtt_send_msg_queue))
        {
            rtos_pop_from_queue(&mqtt_send_msg_queue, &msg_id, 0);
            mqtt_handle_one_message(&msg_id);
        }
    }
    else
    {
        MQTTDisconnect(&client);
        BSD_disconnect(&network);
        rtos_delay_milliseconds(3000);
    }

    MQTTYield(&client, 100);
}

}
int paho_mqtt_start(void)
{
static unsigned char sendbuf[2048];
static unsigned char readbuf[4096];
int rc;

NetworkInit(&network);

rc = NetworkConnect(&network, MQTT_HOST, MQTT_PORT);
if (rc != 0)
    return -1;

g_mqtt_state = MQTT_STATE_TCP_CONNECTED;

MQTTClientInit(&client, &network, 3000, sendbuf, sizeof(sendbuf), readbuf, sizeof(readbuf));

MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer;
connectData.MQTTVersion = 4;
connectData.clientID.cstring = MQTT_CLIENTID;
connectData.cleansession = 0;
connectData.keepAliveInterval = 60;
client.defaultMessageHandler = messageArrived;

printf("--------------------%s\n", MQTT_CLIENTID);
printf("--------------------%s\n", SUBCRIBE_TOPIC);
printf("--------------------%s\n", PUBLISH_TOPIC);

rc = paho_MQTTConnect(&client, &connectData);
if (rc != 0)
    return -2;

g_mqtt_state = MQTT_STATE_MQTT_CONNECTED;

// activate_msg_enter_queue();

rc = MQTTSubscribe(&client, SUBCRIBE_TOPIC, QOS2, messageArrived);
if (rc != 0)
    return -3;

g_mqtt_state = MQTT_STATE_READY;

if (rc != 0)
{
    BSD_disconnect(&network);
    return -2;
}

printf("MQTT connected & subscribed\n");
return 0;

}
6、另外提供一个信息我们现在换成使用emqx cloud的云服务，设备能够正常消费qos2的消息，但是设备也会重连但是不影响消费消息。

zhongwencool · 2026 年2 月 28 日 06:17

根因基本能定了：这是客户端会话实现问题，不是 EMQX 侧参数不够。
你给的验证已经很关键：cleansession=1 时正常、cleansession=0 时异常，而且 paho.mqtt.embedded-c 这边没有 inflight 持久化。这个组合下，断线重连后 QoS2 四步状态丢失，packet_id 很容易被重用，随后就会出现你看到的“消费失败/会话异常”。
再补一个点：你贴的这段 trace 是 PUBLISH(Q1) -> PUBACK，这是 QoS1 方向日志，不是 QoS2 的 PUBREC/PUBREL/PUBCOMP 流程日志。
先按这个方案落地：

如果业务允许，直接用 cleansession=1 + QoS1（你已验证可用）。
如果必须 cleansession=0 + QoS2，就要在设备端实现会话状态持久化：至少持久化未完成 QoS2 事务和 packet_id 分配状态，重连后继续事务，不能从干净状态重新分配。
失败重连路径里，确保不会把“未完成 QoS2 状态”直接丢掉再复用旧 packet_id。
要把这个问题彻底钉死，再贴一段下行 QoS2完整链路文本日志（同一个 packet_id）：PUBLISH(Q2)、PUBREC、PUBREL、PUBCOMP，以及断线原因行（keepalive_timeout / takenover / tcp_closed）。

s264 · 2026 年3 月 3 日 01:43

设备串口打印日志：[09:33:57.230]收←◆(868856):[MQTT PUBLISH] qos=2 dup=0
(868856):packetid=32849
(868856):Message arrived: {“data”:“{"sn":"A0126010129","volumeSize":2,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}
petCharacter = 2
sn = A0126010129

[09:33:59.005]收←◆(870630):[MQTT PUBLISH] qos=2 dup=0
(870630):packetid=32850
(870630):Message arrived: {“data”:“{"sn":"A0126010129","volumeSize":2MQTT reconnecting…,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}
MQTTYield error=-1
petCharacter = 2
sn = A0126010129

[09:33:59.059]收←◆--------------------A0126010129
--------------------device/A0126010129
--------------------service/A0126010129

[09:33:59.104]收←◆(870728):[MQTT PUBLISH] qos=2 dup=1
(870728):packetid=32850
(870728):Message arrived: {
“data”: “{"sn":"A0126010129","type":1}”,
“type”: 19
}
鏈煡绫诲瀷: 19

[09:33:59.145]收←◆MQTT connected & subscribedMQTT connected
mqtt服务器打印日志：
2026-03-03T01:33:46.249659+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBREC(Q0, R0, D0, PacketId=116, ReasonCode=0), username: test
2026-03-03T01:33:46.262563+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0, Topic=device/A0126010129, PacketId=32845, Payload(text)={“data”:“{"sn":"A0126010129","volumeSize":2,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}), username: test
2026-03-03T01:33:46.284652+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBREL(Q1, R0, D0, PacketId=116, ReasonCode=0), username: test
2026-03-03T01:33:46.284823+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBCOMP(Q0, R0, D0, PacketId=116, ReasonCode=0), username: test
2026-03-03T01:33:46.319927+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBREC(Q0, R0, D0, PacketId=32845, ReasonCode=0), username: test
2026-03-03T01:33:46.320107+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0, PacketId=32845, ReasonCode=0), username: test
2026-03-03T01:33:46.356399+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0, PacketId=32845, ReasonCode=0), username: test
2026-03-03T01:33:46.359485+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0, Topic=device/A0126010129, PacketId=32846, Payload(text)={“data”:“{"sn":"A0126010129","volumeSize":2,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}), username: test
2026-03-03T01:33:46.395145+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBREC(Q0, R0, D0, PacketId=32846, ReasonCode=0), username: test
2026-03-03T01:33:46.395324+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0, PacketId=32846, ReasonCode=0), username: test
2026-03-03T01:33:46.442112+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0, PacketId=32846, ReasonCode=0), username: test
2026-03-03T01:33:54.006529+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0, Topic=device/A0126010129, PacketId=32847, Payload(text)={“data”:“{"sn":"A0126010129","volumeSize":2,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}), username: test
2026-03-03T01:33:54.044717+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBREC(Q0, R0, D0, PacketId=32847, ReasonCode=0), username: test
2026-03-03T01:33:54.044946+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0, PacketId=32847, ReasonCode=0), username: test
2026-03-03T01:33:54.079705+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0, PacketId=32847, ReasonCode=0), username: test
2026-03-03T01:33:55.943512+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0, Topic=device/A0126010129, PacketId=32848, Payload(text)={“data”:“{"sn":"A0126010129","volumeSize":2,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}), username: test
2026-03-03T01:33:55.979527+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBREC(Q0, R0, D0, PacketId=32848, ReasonCode=0), username: test
2026-03-03T01:33:55.979697+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0, PacketId=32848, ReasonCode=0), username: test
2026-03-03T01:33:56.023535+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0, PacketId=32848, ReasonCode=0), username: test
2026-03-03T01:33:57.174516+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0, Topic=device/A0126010129, PacketId=32849, Payload(text)={“data”:“{"sn":"A0126010129","volumeSize":2,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}), username: test
2026-03-03T01:33:57.215586+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBREC(Q0, R0, D0, PacketId=32849, ReasonCode=0), username: test
2026-03-03T01:33:57.215781+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0, PacketId=32849, ReasonCode=0), username: test
2026-03-03T01:33:57.250830+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0, PacketId=32849, ReasonCode=0), username: test
2026-03-03T01:33:58.951503+00:00 [MQTT] A0126010129@123.7:45975 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0, Topic=device/A0126010129, PacketId=32850, Payload(text)={“data”:“{"sn":"A0126010129","volumeSize":2,"volumeTime":"2026-03-03 09:32:03"}”,“type”:7}), username: test
2026-03-03T01:33:58.990606+00:00 [SOCKET] A0126010129@123.7:45975 msg: emqx_connection_terminated, reason: {shutdown,tcp_closed}, username: test
2026-03-03T01:33:59.045699+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0, ClientId=A0126010129, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=test, Password=******), username: test
2026-03-03T01:33:59.045906+00:00 [AUTHN] A0126010129@123.7:45242 msg: authenticator_result, authenticator: password_based:built_in_database, result: {ok,#{is_superuser => false}}, username: test
2026-03-03T01:33:59.045983+00:00 [AUTHN] A0126010129@123.7:45242 msg: authentication_result, reason: chain_result, result: {stop,{ok,#{is_superuser => false}}}, username: test
2026-03-03T01:33:59.048177+00:00 [warning] clientid: A0126010129, msg: emqx_persistent_session_ds_replay_inconsistency, peername: 123.7:45242, username: test, expected: {srs,<<“3”>>,32,#{1 => 61,2 => <<“3”>>,3 => #{1 => 2,2 => 32,3 => {it,<<35,95,245,127,231,14,171,176>>,1772500714706999,<<>>,}}},#{1 => 61,2 => <<“3”>>,3 => #{1 => 2,2 => 32,3 => {it,<<35,95,245,127,231,14,171,176>>,1772501638850000,<<>>,}}},1,0,81,0,82,false,1}, got: {srs,<<“3”>>,32,#{1 => 61,2 => <<“3”>>,3 => #{1 => 2,2 => 32,3 => {it,<<35,95,245,127,231,14,171,176>>,1772500714706999,<<>>,}}},#{1 => 61,2 => <<“3”>>,3 => #{1 => 2,2 => 32,3 => {it,<<35,95,245,127,231,14,171,176>>,1772500795455000,<<>>,}}},1,0,81,0,82,false,1}
2026-03-03T01:33:59.048902+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0, AckFlags=1, ReasonCode=0), username: test
2026-03-03T01:33:59.049563+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D1, Topic=device/A0126010129, PacketId=32850, Payload(text)={
“data”: “{"sn":"A0126010129","type":1}”,
“type”: 19
}), username: test
2026-03-03T01:33:59.089869+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_received, packet: SUBSCRIBE(Q1, R0, D0, PacketId=2 TopicFilters=[device/A0126010129(#{nl => 0,qos => 2,rap => 0,rh => 0})]), username: test
2026-03-03T01:33:59.090142+00:00 [AUTHZ] A0126010129@123.7:45242 msg: authorization_module_ignore, action: SUBSCRIBE(Q2), authorize_type: client_info, module: emqx_authz_client_info, topic: device/A0126010129, username: test
2026-03-03T01:33:59.090251+00:00 [AUTHZ] A0126010129@123.7:45242 msg: authorization_matched_allow, action: SUBSCRIBE(Q2), authorize_type: file, module: emqx_authz_file, topic: device/A0126010129, username: test
2026-03-03T01:33:59.091089+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_sent, packet: SUBACK(Q0, R0, D0, PacketId=2, ReasonCodes=[2]), username: test
2026-03-03T01:33:59.131105+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_received, packet: PUBREC(Q0, R0, D0, PacketId=32850, ReasonCode=0), username: test
2026-03-03T01:33:59.131311+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0, PacketId=32850, ReasonCode=0), username: test
2026-03-03T01:33:59.332927+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0, PacketId=32850, ReasonCode=0), username: test
2026-03-03T01:34:01.994118+00:00 [MQTT] A0126010129@123.7:45242 msg: mqtt_packet_received, packet: PUBLISH(Q2, R0, D0, Topic=service/A0126010129, PacketId=3, Payload(text)={“type”:5,“data”:{“type”:“V0002”,“message”:“123”,“sn”:“A0126010129”,“emotion”:0,“takeTime”:708}}), username: test
2026-03-03T01:34:01.994367+00:00 [AUTHZ] A0126010129@123.7:45242 msg: authorization_module_ignore, action: PUBLISH(Q2,R0), authorize_type: client_info, module: emqx_authz_client_info, topic: service/A0126010129, username: test
2026-03-03T01:34:01.994432+00:00 [AUTHZ] A0126010129@123.7:45242 msg: authorization_matched_allow, action: PUBLISH(Q2,R0), authorize_type: file, module: emqx_authz_file, topic: service/A0126010129, username: test
2026-03-03T01:34:01.994520+00:00 [PUBLISH] A0126010129@123.7:45242 msg: publish_to, payload_encode: text, topic: service/A0126010129, username: test, payload: {“type”:5,“data”:{“type”:“V0002”,“message”:“123。”,“sn”:“A0126010129”,“emotion”:0,“takeTime”:708}}
是应该是 packet_id=32850 目前我们的业务是要 cleansession=0 + QoS2

zhongwencool · 2026 年3 月 3 日 02:23

原因已经比较明确：是客户端在 QoS2 处理中断线，重连后触发重投（dup=1），你的业务层又没有把重复包/未完成事务处理好，所以看起来像“消费失败”。
你贴的这段关键证据：MQTTYield error=-1 紧接着重连，然后同一个 packetid=32850 以 dup=1 再次下发。这是 QoS2 的正常重投行为，不是 broker 乱发。
另外 type=19 被打印成“未知类型”，这是你应用协议解析分支没覆盖，不是 MQTT 协议层错误。
可以这个顺序改：

接收回调里不要做耗时逻辑（串口打印/JSON 解析/业务处理都放到队列），回调只做“入队+快速返回”。
对 dup=1 做幂等处理：按 sn + packetid（或业务消息唯一键）去重，重复包直接 ACK 并丢弃业务重复执行。
如果继续用 cleanSession=0 + QoS2，必须自己持久化未完成 QoS2 状态（至少 packet_id 分配状态 + 未完成事务）。
如果短期先要稳定，先切到 cleanSession=1 + QoS1（你已经验证这条是稳定的）。
建议你直接补两段日志再确认：

# 设备侧（同一个 packetid）
PUBLISH(Q2) -> PUBREC -> PUBREL -> PUBCOMP
# EMQX 侧（同一时间窗）
断开原因行（tcp_closed / keepalive_timeout / takenover）

重点关注你的 messageArrived 回调和重连逻辑代码（尤其是断线后 packet_id/会话状态怎么处理）。

zhongwencool · 2026 年3 月 3 日 02:28

github.com/emqx/emqx

QoS 2 PUBLISH Rejected with Packet Identifier In Use (RC 145) Immediately After Previous QoS 2 Flow Completion

opened 08:35AM - 17 Apr 25 UTC

closed 07:29AM - 08 May 25 UTC

zhongwencool

BUG

### What happened? From : https://askemq.com/t/topic/11275 ## Environment: EMQ…X Version: 5.6.1 (Open Source) Client Library: spring-integration-mqtt v5.5.20 Underlying MQTT Client: org.eclipse.paho.mqttv5.client v1.2.5 MQTT Version: 5 Client Configuration: CleanStart=false (though the error occurs within a single connection instance before takeover) ## Description: We are observing an issue where EMQX rejects a QoS 2 PUBLISH packet with Reason Code 145 (Packet Identifier in use) immediately after successfully completing a prior QoS 2 message exchange using a different Packet ID within the same client connection. ### What did you expect to happen? After a QoS 2 message flow for a given Packet ID (e.g., 888) is fully completed (signified by the broker sending PUBCOMP), that Packet ID should be considered free. When the client sends a new PUBLISH message with a different, previously unused (or fully completed) Packet ID (e.g., 889), the broker should accept it and initiate the QoS 2 flow by responding with a PUBREC with Reason Code 0 (Success), assuming no other constraints are violated. ### How can we reproduce it (as minimally and precisely as possible)? _No response_ ### Anything else we need to know? ## Additional Context: The client application appears to be sending messages in a loop, potentially leading to rapid sequential use of Packet IDs. The user provided EMQX trace logs and Wireshark captures in the original forum post (links/attachments would ideally be included here if available). The user mentioned setting a client message rate limit (listeners.tcp.default.messages_rate = "100/s"). ### EMQX version <details> ```console $ ./bin/emqx_ctl broker # paste output here ``` </details> ### OS version <details> ```console # On Linux: $ cat /etc/os-release # paste output here $ uname -a # paste output here ``` </details> <img width="1637" alt="Image" src="https://github.com/user-attachments/assets/322f7d59-d437-4d68-9e83-bc67ad80b31c" /> ### Log files [fu_emqx@127.0.0.1-trace_producer_client_local1_11_2025-04-16.log](https://github.com/user-attachments/files/19791127/fu_emqx%40127.0.0.1-trace_producer_client_local1_11_2025-04-16.log) [pcap-trace.zip](https://github.com/user-attachments/files/19791134/v5-5.zip) **Notes**: These two logs did not occur in the same time

据我所知，paho SDK 不提供客户端会话和消息的状态持久化（例如写入磁盘），所以程序重启的情况，会导致 qos2 消息状态丢失。
EMQX 保留了这些状态，客户端重连后，重新使用了之前没发完QoS2的 packet ID，EMQX 就会拒绝这个 PUBLISH 消息。

建议不要使用 QoS2。

s264 · 2026 年3 月 3 日 02:52

您好如果我们使用qos1+CleanSession=false 然后设备自己做业务消息的幂等判定，这样只是解决消息消费的失败的问题还是说能够同步解决设备重连的问题。如果不能解决重连的问题，有没有其他的排查方向或者解决方案来解决重连的问题，因为qos1也会进行2次握手我担心经常重连也会影响消息消息出现类似的问题而且在网络稳定的情况下经常重连也不应该发生。

zhongwencool · 2026 年3 月 3 日 03:04

QoS1 + CleanSession=false + 业务幂等，只能绕开 QoS2 会话状态丢失导致的消费失败，不能根治【频繁重连】。重连要单独排查链路、线程阻塞和断开原因。
你现在的日志里已经有信号：MQTTYield error=-1 后立刻重连，这通常是 socket 读写失败或网络线程被阻塞，不是 QoS1/QoS2 本身引起。

s264 · 2026 年3 月 3 日 03:42

如果解决不了设备频繁重连的问题，消费qos1消息的时候会不会出现同样消费qos2消息失败的情况会不会造成其它异常情况例如消息消费失败后丢失等等影响业务的情况。也就是QoS1 + CleanSession=false + 业务幂等+设置频繁重连会不会影响业务。

zhongwencool · 2026 年3 月 3 日 04:51

会影响业务，但可控：QoS1 + CleanSession=false + 业务幂等能规避你现在这个 QoS2 会话状态丢失问题，不能消除频繁重连本身。

QoS1 不会出现 QoS2 那种 packet_id in use 的四步状态冲突。
频繁重连时，QoS1 典型表现是重复投递和时延抖动，不是协议层必然丢消息。
是否丢业务消息，取决于你应用层是否做到重试、幂等、离线期间补偿，不是 QoS1 单独决定。

在你这个场景，建议按这个组合来做：

继续用 QoS1 + CleanSession=false。
设备侧幂等键用 sn + 业务消息唯一ID，不要只靠 packet_id。
收包回调只做入队，业务处理放工作线程，避免 MQTTYield 被阻塞再次触发 error=-1。
重连做指数退避（1s/2s/4s 递增，上限 30s），不要紧循环硬重连。

QoS1 + 幂等可以先把消费异常风险压住；频繁重连如果不治，仍会带来重复消费、延迟和瞬时堆积，需要单独定位查断线重连的原因。
你说是在办公室的 wifi 上，这 TCP应该是很稳定的才对。我也不明白。