客户端连接每30秒断开，3秒后重新连接

EMQX
,
1

EMQX 版本： V5.1.5
服务器OS： CentOS 7.9
客户端A： MQTTX
客户端B： php7.2

业务流程：

  1. 配置规则， 捕捉 客户端A的连接成功和断开事件；
  2. 根据捕捉的事件，重发消息, topic为 status/change
  3. 客户端B 订阅 status/change，将状态变更写入数据库。

问题： 客户端A连接成功后， 然后每30秒总是执行一次断开连接和重新连接， 为什么会有这样的定期断开重连动作？

附件有两个日志：

  1. 客户端B打印的修改数据库的日志。
  2. EMQX 的debug日志

2023-08-23T20:00:36.174748+00:00 [debug] msg: raw_bin_received, mfa: emqx_connection:when_bytes_in/3, line: 779, peername: 86.242.108.98:53845, bin: 103F00044D51545404CE003C00043130303100167265706F7274656E642F313030312F73746F707065640006627965627965000376707200083132333435364163, size: 65, type: hex
2023-08-23T20:00:36.175575+00:00 [debug] msg: mqtt_packet_received, mfa: emqx_channel:handle_in/2, line: 335, peername: 86.242.108.98:53845, clientid: 991, packet: CONNECT(Q0, R0, D0, ClientId=1001, ProtoName=MQTT, ProtoVsn=4, CleanStart=true, KeepAlive=60, Username=clientA, Password=******, Will(Q1, R0, Topic=reportend/1001/stopped Payload=byebye)), tag: MQTT
2023-08-23T20:00:36.176311+00:00 [debug] msg: mysql_connector_received, mfa: emqx_mysql:on_sql_query/6, line: 455, peername: 86.242.108.98:53845, clientid: 991, connector: <<“emqx_authn_mysql:1”>>, sql: emqx_authn_mysql, state: #{batch_inserts => #{},batch_params_tokens => #{},params_tokens => #{emqx_authn_mysql => []},pool_name => <<“emqx_authn_mysql:1”>>,prepare_statement => #{emqx_authn_mysql => <<“SELECT secret_key as password_hash FROM clientA_printer where dev_id = ? LIMIT 1”>>}}, tag: QUERY
2023-08-23T20:00:36.177488+00:00 [debug] msg: authenticator_result, mfa: emqx_authentication:authenticate_with_provider/2, line: 699, peername: 86.242.108.98:53845, clientid: 991, authenticator: <<“password_based:mysql”>>, result: {ok,#{is_superuser => false}}, tag: AUTHN
2023-08-23T20:00:36.177842+00:00 [debug] msg: authentication_result, mfa: emqx_authentication:authenticate/2, line: 247, peername: 86.242.108.98:53845, clientid: 991, reason: chain_result, result: {stop,{ok,#{is_superuser => false}}}, tag: AUTHN
2023-08-23T20:00:36.178419+00:00 [debug] msg: republish_message, mfa: emqx_rule_actions:republish/3, line: 123, peername: 86.242.108.98:53845, clientid: 991, topic: status/connect, flags: #{retain => false}, payload: {“username”:“clientA”,“clientid”:“1001”}, pub_props: #{‘User-Property’ => #{}}, tag: RULE
2023-08-23T20:00:36.178787+00:00 [debug] msg: publish_to, mfa: emqx_trace:publish/1, line: 73, peername: 86.242.108.98:53845, clientid: 991, topic: status/connect, payload: {“username”:“clientA”,“clientid”:“1001”}, tag: PUBLISH
2023-08-23T20:00:36.179066+00:00 [debug] msg: insert_channel_info, mfa: emqx_cm:insert_channel_info/3, line: 159, peername: 86.242.108.98:53845, clientid: 991
2023-08-23T20:00:36.179244+00:00 [debug] msg: mqtt_packet_sent, mfa: emqx_connection:serialize_and_inc_stats_fun/1, line: 878, peername: 86.242.108.98:53845, clientid: 991, packet: CONNACK(Q0, R0, D0, AckFlags=0, ReasonCode=0), tag: MQTT
2023-08-23T20:00:41.427803+00:00 [debug] msg: raw_bin_received, mfa: emqx_connection:when_bytes_in/3, line: 779, peername: 86.242.108.98:53428, clientid: SWOB035721684, bin: C000, size: 2, type: hex
2023-08-23T20:00:41.428432+00:00 [debug] msg: mqtt_packet_received, mfa: emqx_connection:handle_msg/2, line: 575, peername: 86.242.108.98:53428, clientid: SWOB035721684, packet: PINGREQ(Q0, R0, D0), tag: MQTT
2023-08-23T20:00:41.428756+00:00 [debug] msg: mqtt_packet_sent, mfa: emqx_connection:serialize_and_inc_stats_fun/1, line: 878, peername: 86.242.108.98:53428, clientid: SWOB035721684, packet: PINGRESP(Q0, R0, D0), tag: MQTT
2023-08-23T20:00:43.876837+00:00 [debug] event=send_data_succeeded socket=“#Port<0.211>”
2023-08-23T20:00:43.877180+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.211>”
2023-08-23T20:00:43.877685+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.212>” peer=“127.0.0.1:38570” action=ignore
2023-08-23T20:00:49.844766+00:00 [debug] event=send_data_succeeded socket=“#Port<0.216>”
2023-08-23T20:00:49.845012+00:00 [debug] message=ping event=transmission_succeeded driver=tcp socket=“#Port<0.216>”
2023-08-23T20:00:49.846316+00:00 [debug] event=ping_received driver=tcp socket=“#Port<0.217>” peer=“127.0.0.1:38574” action=ignore

2023-08-23T20:16:05.886091+00:00 [debug] msg: raw_bin_received, mfa: emqx_connection:when_bytes_in/3, line: 779, peername: 86.242.108.98:55267, bin: 103F00044D51545404CE003C00043130303100167265706F7274656E642F313030312F73746F707065640006627965627965000376707200083132333435364163, size: 65, type: hex
2023-08-23T20:16:05.886796+00:00 [debug] msg: mqtt_packet_received, mfa: emqx_channel:handle_in/2, line: 335, peername: 86.242.108.98:55267, clientid: 991, packet: CONNECT(Q0, R0, D0, ClientId=1001, ProtoName=MQTT, ProtoVsn=4, CleanStart=true, KeepAlive=60, Username=clientA, Password=******, Will(Q1, R0, Topic=reportend/1001/stopped Payload=byebye)), tag: MQTT
2023-08-23T20:16:05.887413+00:00 [debug] msg: mysql_connector_received, mfa: emqx_mysql:on_sql_query/6, line: 455, peername: 86.242.108.98:55267, clientid: 991, connector: <<“emqx_authn_mysql:1”>>, sql: emqx_authn_mysql, state: #{batch_inserts => #{},batch_params_tokens => #{},params_tokens => #{emqx_authn_mysql => []},pool_name => <<“emqx_authn_mysql:1”>>,prepare_statement => #{emqx_authn_mysql => <<“SELECT secret_key as password_hash FROM clientA_printer where dev_id = ? LIMIT 1”>>}}, tag: QUERY
2023-08-23T20:16:05.889396+00:00 [debug] msg: authenticator_result, mfa: emqx_authentication:authenticate_with_provider/2, line: 699, peername: 86.242.108.98:55267, clientid: 991, authenticator: <<“password_based:mysql”>>, result: {ok,#{is_superuser => false}}, tag: AUTHN
2023-08-23T20:16:05.889807+00:00 [debug] msg: authentication_result, mfa: emqx_authentication:authenticate/2, line: 247, peername: 86.242.108.98:55267, clientid: 991, reason: chain_result, result: {stop,{ok,#{is_superuser => false}}}, tag: AUTHN
2023-08-23T20:16:05.890464+00:00 [debug] msg: republish_message, mfa: emqx_rule_actions:republish/3, line: 123, peername: 86.242.108.98:55267, clientid: 991, topic: status/connect, flags: #{retain => false}, payload: {“username”:“clientA”,“clientid”:“1001”}, pub_props: #{‘User-Property’ => #{}}, tag: RULE
2023-08-23T20:16:05.890908+00:00 [debug] msg: publish_to, mfa: emqx_trace:publish/1, line: 73, peername: 86.242.108.98:55267, clientid: 991, topic: status/connect, payload: {“username”:“clientA”,“clientid”:“1001”}, tag: PUBLISH
2023-08-23T20:16:05.891307+00:00 [debug] msg: insert_channel_info, mfa: emqx_cm:insert_channel_info/3, line: 159, peername: 86.242.108.98:55267, clientid: 991
2023-08-23T20:16:05.891599+00:00 [debug] msg: mqtt_packet_sent, mfa: emqx_connection:serialize_and_inc_stats_fun/1, line: 878, peername: 86.242.108.98:55267, clientid: 991, packet: CONNACK(Q0, R0, D0, AckFlags=0, ReasonCode=0), tag: MQTT
2023-08-23T20:16:10.395759+00:00 [debug] msg: raw_bin_received, mfa: emqx_connection:when_bytes_in/3, line: 779, peername: 86.242.108.98:54175, clientid: SWOB035721684, bin: C000, size: 2, type: hex
2023-08-23T20:16:10.396289+00:00 [debug] msg: mqtt_packet_received, mfa: emqx_connection:handle_msg/2, line: 575, peername: 86.242.108.98:54175, clientid: SWOB035721684, packet: PINGREQ(Q0, R0, D0), tag: MQTT
2023-08-23T20:16:10.396604+00:00 [debug] msg: mqtt_packet_sent, mfa: emqx_connection:serialize_and_inc_stats_fun/1, line: 878, peername: 86.242.108.98:54175, clientid: SWOB035721684, packet: PINGRESP(Q0, R0, D0), tag: MQTT

image
image1440×584 96.5 KB

2

看一下这个命令的输出结果，里面会有掉线原因的统计：emqx ctl listeners

3

按照你的建议，操作如下：

  1. 重启emqx，清空emqx ctl listeners 历史数据
  2. 重新连接客户端A和B，输入如下。
    结果：通过规则 设置，捕捉到每30秒断开和重连。但是emqx ctl listeners里面没有这种断开和重连的记录。
    image
    image890×842 114 KB

    image
    image1460×584 61.9 KB
4
  • 从 emqx ctl listeners 命令中，只看到一个 1883 端口的掉线记录(shutdown_count 字段里有一个 tcp_closed)，你确定你的客户端走的是 8883 吗？
  • 如果是异常掉线，会触发你客户端设置的 will topic，你能收到 reportend/1001/stopped 主题的消息吗？
5

1 . 确定客户端A走的时8883端口。
image

  1. 规则设置中，转发了connect和disconnect的所有参数，打印如下：

    image
    image2518×1020 385 KB

  2. 打印emqx broker捕捉到的客户端B的debug日志如下：

    image
    image1920×720 246 KB

6

几点分析，不知道对不对：

  1. 应该不是真正的断开和重新连接。 debug日志中没有看出来。
  2. 断开和重新连接是通过规则捕获的， 捕获后以消息重发的方式发给了应用打印。 从打印的消息看，是因为 keep-alive timeout。
  3. 对比锻炼消息和debug日志，和每次心跳的时间吻合。
  4. 找到之前的一个类似话题，但是没有最终结论 ( 客户端连接broker后频繁离线，客户端数据上报频率5分钟一次，心跳从60s到420s均尝试过，日志显示“socket_force_closed, reason: keepalive_timeout”，“emqx_connection_terminated, reason: {shutdown,keepalive_timeout}” )。可能是MQTTX的心跳问题。可以肯定的一点是网络没有问题。
    MQTTX这样设置有问题吗？
    image
    image1574×1210 103 KB