java写了一个mqtt去订阅数据，消费时间可能有点长，消费几条数据后，服务器中的日志先发出警告，然后重新订阅和取消订阅，然后出现断联和协议错误等信息，如何让客户端稳定消化需要长时间处理的数据？

YeYu · 2024 年10 月 25 日 10:24

问题求助：

emqx 5.0.7 系统 centos7，使用docker容器部署

2024-10-25T07:21:44+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0),Topic=/iotcom/query/out, PacketId=7Payload=
2024-10-25T07:23:11+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_received, packet: PUBCOMP(Q0, R0, D0),PacketId=6, ReasonCode=0
2024-10-25T07:23:11.130599+00:00 [warning] clientid: meter127.0.0.1, line: 495, mfa: emqx_channel:handle_in/2, msg: pubcomp_packetId_not_found, packetId: 6, peername: 192.168.30.249:47804
2024-10-25T07:23:14+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0),PacketId=7, ReasonCode=0
2024-10-25T07:23:44+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0),PacketId=7, ReasonCode=0
2024-10-25T07:24:11+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_received, packet: PINGREQ(Q0, R0, D0)
2024-10-25T07:24:11+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_sent, packet: PINGRESP(Q0, R0, D0)
2024-10-25T07:24:14+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0),PacketId=7, ReasonCode=0
2024-10-25T07:24:44+00:00 [MQTT] meter127.0.0.1@192.168.30.249:47804 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0),PacketId=7, ReasonCode=0
2024-10-25T07:27:42+00:00 [MQTT] meter127.0.0.1@192.168.30.249:33942 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:27:42+00:00 [SUBSCRIBE] meter127.0.0.1@192.168.30.249:33942 msg: subscribe, sub_id: meter127.0.0.1, sub_opts: [nl: 0, qos: 2, rap: 0, rh: 0, sub_props: ], topic: /iotcom/query/out
2024-10-25T07:27:42+00:00 [UNSUBSCRIBE] meter127.0.0.1@192.168.30.249:47804 msg: unsubscribe, sub_opts: [nl: 0, qos: 2, rap: 0, rh: 0, sub_props: , subid: meter127.0.0.1], topic: /iotcom/query/out
2024-10-25T07:27:42.450891+00:00 [warning] clientid: meter127.0.0.1, line: 821, mfa: emqx_session:redispatch_shared_messages/1, msg: qos2_lost_no_redispatch, peername: 192.168.30.249:47804
2024-10-25T07:27:42.451168+00:00 [debug] client_id: <<“meter127.0.0.1”>>, clientid: meter127.0.0.1, line: 155, mfa: emqx_cm:insert_channel_info/3, msg: insert_channel_info, peername: 192.168.30.249:33942
2024-10-25T07:27:42+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:47804 msg: emqx_connection_terminated, reason: {shutdown,takenover}
2024-10-25T07:27:42+00:00 [MQTT] meter127.0.0.1@192.168.30.249:33942 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=1, ReasonCode=0
2024-10-25T07:27:42.451660+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:47804, reason: {shutdown,takenover}
2024-10-25T07:27:42+00:00 [MQTT] meter127.0.0.1@192.168.30.249:33942 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0),PacketId=7, ReasonCode=0
2024-10-25T07:27:42+00:00 [MQTT] meter127.0.0.1@192.168.30.249:33942 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:27:42+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:33942 msg: socket_force_closed, reason: protocol_error
2024-10-25T07:34:07+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50844 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQIsdp, ProtoVsn=3, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:07+00:00 [SUBSCRIBE] meter127.0.0.1@192.168.30.249:50844 msg: subscribe, sub_id: meter127.0.0.1, sub_opts: [nl: 0, qos: 2, rap: 0, rh: 0, sub_props: ], topic: /iotcom/query/out
2024-10-25T07:34:07+00:00 [UNSUBSCRIBE] meter127.0.0.1@192.168.30.249:33942 msg: unsubscribe, sub_opts: [nl: 0, qos: 2, rap: 0, rh: 0, sub_props: , subid: meter127.0.0.1], topic: /iotcom/query/out
2024-10-25T07:34:07.742332+00:00 [warning] clientid: meter127.0.0.1, line: 821, mfa: emqx_session:redispatch_shared_messages/1, msg: qos2_lost_no_redispatch, peername: 192.168.30.249:33942
2024-10-25T07:34:07.742631+00:00 [debug] client_id: <<“meter127.0.0.1”>>, clientid: meter127.0.0.1, line: 155, mfa: emqx_cm:insert_channel_info/3, msg: insert_channel_info, peername: 192.168.30.249:50844
2024-10-25T07:34:07+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50844 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=1, ReasonCode=0
2024-10-25T07:34:07+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:33942 msg: emqx_connection_terminated, reason: {shutdown,takenover}
2024-10-25T07:34:07.743145+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:33942, reason: {shutdown,takenover}
2024-10-25T07:34:07+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50844 msg: mqtt_packet_sent, packet: PUBREL(Q1, R0, D0),PacketId=7, ReasonCode=0
2024-10-25T07:34:07+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50844 msg: mqtt_packet_sent, packet: PUBLISH(Q2, R0, D0),Topic=/iotcom/query/out, PacketId=8Payload=
2024-10-25T07:34:07+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50844 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:07+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:50844 msg: socket_force_closed, reason: protocol_error
2024-10-25T07:34:07+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50844 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQIsdp, ProtoVsn=3, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:07+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50844 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQIsdp, ProtoVsn=3, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:10+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50850 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:10+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50850 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=0, ReasonCode=2
2024-10-25T07:34:10+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:50850 msg: emqx_connection_terminated, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:10.038486+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:50850, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:10+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50864 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:10+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50864 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=0, ReasonCode=2
2024-10-25T07:34:10+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:50864 msg: emqx_connection_terminated, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:10.337312+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:50864, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:14+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50866 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQIsdp, ProtoVsn=3, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:14+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50866 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=0, ReasonCode=2
2024-10-25T07:34:14+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:50866 msg: emqx_connection_terminated, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:14.555507+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:50866, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:14+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50878 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:14+00:00 [MQTT] meter127.0.0.1@192.168.30.249:50878 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=0, ReasonCode=2
2024-10-25T07:34:14+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:50878 msg: emqx_connection_terminated, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:14.844714+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:50878, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:22.825029+00:00 [warning] action: {takeover,‘begin’}, clientid: meter127.0.0.1, line: 480, mfa: emqx_cm:request_stepdown/3, msg: session_stepdown_request_timeout, peername: 192.168.30.249:50844, pid: <0.11756.205>, stale_channel: [{status,running},{message_queue_len,8},{current_stacktrace,[{emqx_cm,stale_channel_info,1,[{file,“emqx_cm.erl”},{line,510}]},{emqx_cm,request_stepdown,3,[{file,“emqx_cm.erl”},{line,481}]},{emqx_cm,do_takeover_session,2,[{file,“emqx_cm.erl”},{line,421}]},{emqx_cm,takeover_session,2,[{file,“emqx_cm.erl”},{line,402}]},{emqx_cm,‘-open_session/3-fun-2-’,5,[{file,“emqx_cm.erl”},{line,291}]},{emqx_cm_locker,trans,3,[{file,“emqx_cm_locker.erl”},{line,51}]},{emqx_channel,process_connect,2,[{file,“emqx_channel.erl”},{line,611}]},{emqx_connection,with_channel,3,[{file,“emqx_connection.erl”},{line,790}]},{emqx_connection,process_msg,2,[{file,“emqx_connection.erl”},{line,466}]},{emqx_connection,process_msg,2,[{file,“emqx_connection.erl”},{line,472}]},{emqx_connection,handle_recv,3,[{file,“emqx_connection.erl”},{line,428}]},{proc_lib,wake_up,3,[{file,“proc_lib.erl”},{line,236}]}]}]
2024-10-25T07:34:23+00:00 [MQTT] meter127.0.0.1@192.168.30.249:42934 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQIsdp, ProtoVsn=3, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:23+00:00 [MQTT] meter127.0.0.1@192.168.30.249:42934 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=0, ReasonCode=2
2024-10-25T07:34:23+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:42934 msg: emqx_connection_terminated, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:23.070590+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:42934, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:23+00:00 [MQTT] meter127.0.0.1@192.168.30.249:42946 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:23+00:00 [MQTT] meter127.0.0.1@192.168.30.249:42946 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=0, ReasonCode=2
2024-10-25T07:34:23+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:42946 msg: emqx_connection_terminated, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:23.352179+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:42946, reason: {shutdown,client_identifier_not_valid}
2024-10-25T07:34:39+00:00 [MQTT] meter127.0.0.1@192.168.30.249:59756 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQIsdp, ProtoVsn=3, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:39.592952+00:00 [debug] client_id: <<“meter127.0.0.1”>>, clientid: meter127.0.0.1, line: 155, mfa: emqx_cm:insert_channel_info/3, msg: insert_channel_info, peername: 192.168.30.249:59756
2024-10-25T07:34:39+00:00 [MQTT] meter127.0.0.1@192.168.30.249:59756 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=0, ReasonCode=0
2024-10-25T07:34:39+00:00 [MQTT] meter127.0.0.1@192.168.30.249:59756 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:39+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:59756 msg: socket_force_closed, reason: protocol_error
2024-10-25T07:34:41+00:00 [MQTT] meter127.0.0.1@192.168.30.249:59760 msg: mqtt_packet_received, packet: CONNECT(Q0, R0, D0),ClientId=meter127.0.0.1, ProtoName=MQTT, ProtoVsn=4, CleanStart=false, KeepAlive=60, Username=admin, Password=******
2024-10-25T07:34:41+00:00 [SOCKET] meter127.0.0.1@192.168.30.249:59756 msg: emqx_connection_terminated, reason: {shutdown,takenover}
2024-10-25T07:34:41.965187+00:00 [info] clientid: meter127.0.0.1, line: 636, mfa: emqx_connection:terminate/2, msg: terminate, peername: 192.168.30.249:59756, reason: {shutdown,takenover}
2024-10-25T07:34:41.965316+00:00 [debug] client_id: <<“meter127.0.0.1”>>, clientid: meter127.0.0.1, line: 155, mfa: emqx_cm:insert_channel_info/3, msg: insert_channel_info, peername: 192.168.30.249:59760
2024-10-25T07:34:41+00:00 [MQTT] meter127.0.0.1@192.168.30.249:59760 msg: mqtt_packet_sent, packet: CONNACK(Q0, R0, D0),AckFlags=1, ReasonCode=0

没有长时间稳定运行过，每天会有几条长数据过来，有时候漏推了会连续推过来，每条数据大概需要10-20分钟处理，连续处理数据就会崩溃。

设置过connectionTimeout 和 timeout 参数没有解决问题

rocky · 2024 年10 月 25 日 10:35

处理数据在异步线程里进行处理，不要跟接收消息的线程在一起，否则会堵塞消息接收线程。

zhongwencool · 2024 年10 月 27 日 02:15

这里有完整的参照例子。