后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载导读当 EMQX 中某个客户端连接进程的邮箱队列或堆内存超过force_shutdown.max_mailbox_size/force_shutdown.max_heap_size阈值而被强制关闭时旧版本的关闭报告里只有触发的阈值与实测值运维人员无法得知被关闭的究竟是哪个客户端。本篇文章围绕该修复对应仓库变更记录 fix-18521.en.md完整梳理 EMQX 强制关闭机制的配置项、OOM 判定调用链以及本次为关闭报告新增label字段的底层实现原理帮助读者掌握如何定位与排查此类连接被强制关闭事件。一、问题背景强制关闭报告为何难以排查EMQX 为每个客户端连接维护一个独立的 Erlang 进程。为防止单个失控进程拖垮整个节点EMQX 提供了基于邮箱队列长度与进程堆大小的强制关闭force shutdown机制一旦连接进程的message_queue_len或total_heap_size超过配置阈值该连接会被立即终止以换取系统整体稳定性。在修复之前此类关闭产生的报告只包含触发关闭的原因如mailbox_overflow、proc_heap_too_large、实测值value与阈值max却缺少被关闭进程的身份信息。当多个客户端同时连接、批量触发强制关闭时运维人员无法从日志中分辨出具体是哪个客户端被关闭排障往往需要逐条对照进程 ID 与客户端映射效率极低。本次修复的核心改动见 emqx_utils.erl是在关闭原因中追加一个label字段——对于已完成 CONNECT 的已建立连接label保存的是客户端 IDClient ID对于在 CONNECT 完成之前就被关闭的连接label保存的是监听器名称与对端地址peer address。这样无论是已上线客户端还是握手阶段的新连接都能在关闭报告中得到有效辨识。二、force_shutdown 配置项详解force_shutdown属于监听器/Zone 级别的配置。其字段定义位于 emqx_schema.erl配置项类型默认值说明force_shutdown.enableboolean()true是否启用强制关闭机制标记为IMPORTANCE_NO_DOC即不在默认文档中展示的辅助开关force_shutdown.max_mailbox_sizerange(0, inf)1000连接进程邮箱队列message queue的最大长度超过该长度即触发关闭兼容旧别名max_message_queue_lenforce_shutdown.max_heap_sizewordsize()即字大小如32MB32MB连接进程堆的最大大小超过即触发关闭配置了独立的校验器validate_heap_size/1对应的运行时类型定义在 emqx_types.erl 中称为 OOM 策略oom_policy()-type oom_policy() :: #{ max_mailbox_size non_neg_integer(), max_heap_size non_neg_integer(), enable boolean() }.在连接进程启动时Zone 的force_shutdown配置会被读取并用于调整进程堆大小上限见 emqx_connection.erlShutdownPolicy emqx_config:get_zone_conf(Zone, [force_shutdown]), emqx_utils:tune_heap_size(ShutdownPolicy),其中tune_heap_size/1的实现位于 emqx_utils.erl当enable为false或max_heap_size为0时忽略否则将max_heap_size放大 1.5 倍并写入进程标志max_heap_sizekill true, error_logger true使 Erlang VM 在进程堆溢出时以更强的方式介入。注意这里 1.5 倍溢出窗口是一个运行期保护垫实际对外暴露的配置阈值仍是原始max_heap_size。典型配置示例emqx.conf/ Zone 配置zone.default { force_shutdown { enable true max_mailbox_size 1000 max_heap_size 32MB } }提示max_mailbox_size的语义是邮箱队列长度条数并非字节数max_heap_size以字为单位在 64 位平台上每字 8 字节。三、OOM 判定与关闭报告的生成链路连接进程在每次处理完一批入站消息后都会调用check_oom/3检查自身邮箱与堆是否越限见 emqx_connection.erlcheck_oom(Pubs, Bytes, #state{conf #conf{force_shutdown ShutdownPolicy}}) - case emqx_utils:check_oom(ShutdownPolicy) of {shutdown, #{reason : OomReason} Detail} - ?SLOG(error, Detail#{msg connection_shutdown_due_to_oom}), ?tp_ignore_side_effects_in_prod(check_oom_shutdown, #{ policy ShutdownPolicy, incoming_pubs Pubs, incoming_bytes Bytes, shutdown Detail }), %% triggers terminate/2 callback immediately erlang:exit({shutdown, OomReason}); Result - ?tp(debug, check_oom_ok, #{...}) end.关键调用链如下emqx_utils:check_oom(Policy)读取当前进程的message_queue_len与total_heap_size见 emqx_utils.erl并依次比对两个阈值邮箱长度超限 → 原因mailbox_overflow堆大小超限 → 原因proc_heap_too_large。判定逻辑do_check_oom/2只有在0 Max Val阈值有效且实测值严格大于阈值时才判定越限返回{shutdown, #{reason Reason, value Val, max Max}}。连接进程随即通过erlang:exit({shutdown, OomReason})触发terminate/2回调连接被强制关闭同时输出connection_shutdown_due_to_oom错误日志。修复前的关闭原因形如#{ reason mailbox_overflow, value 1200, max 1000 }此时日志只能看到邮箱长度 1200 超过 1000却不知道是谁的邮箱。四、label 字段的注入机制修复的核心实现本次修复在关闭原因中加入了label字段注入点位于 emqx_utils.erl 的add_proc_label/2%% Add the process label (e.g. {clientid, ClientId}) to the shutdown %% reason so the supervisor report identifies the offending process. add_proc_label(_Pid, ok) - ok; add_proc_label(Pid, {shutdown, Reason}) - case proc_lib:get_label(Pid) of undefined - {shutdown, Reason}; Label - {shutdown, Reason#{label Label}} end.它的工作原理是读取 Erlang OTP 的proc_lib进程标签process label如果该进程已被设置过标签就把标签并入关闭原因如果没有标签则保持原样返回。check_oom/2在完成越限判定后立即调用add_proc_label(Pid, Result)emqx_utils.erl因此只要进程设置了标签关闭报告就一定携带label。那么label的值从哪里来EMQX 在不同连接阶段设置了不同的进程标签4.1 已建立连接label 为客户端 ID当 CONNECT 报文处理完成、连接正式建立时通道层会调用 emqx_channel.erl 的set_log_meta/2set_log_meta(_ConnPkt, #channel{clientinfo #{clientid : ClientId} ClientInfo}) - proc_lib:set_label(ClientId), Username maps:get(username, ClientInfo, undefined), ... emqx_logger:set_proc_metadata([{username, Username}, {tns, Tns}]).这里直接以客户端 ID 作为进程标签。因此凡是完成 CONNECT 的连接一旦后续触发强制关闭报告中的label即为该连接的Client ID运维人员可以直接在日志中按客户端 ID 检索。4.2 CONNECT 完成前的连接label 为监听器与对端地址对于尚未完成 CONNECT 就触发关闭的连接例如握手阶段堆积了过量的报文此时还没有可用的 Client ID。EMQX 在连接进程启动早期即设置另一个标签见 emqx_connection_util.erllabel_process( {Type, ListenerName}, #{peername : Peername} _ConnInfo ) - PeernameStr format_peername(Peername), ListenerId emqx_listeners:listener_id(Type, ListenerName), proc_lib:set_label({ListenerId, PeernameStr}), emqx_logger:set_proc_metadata(#{ listener ListenerId, peername PeernameStr }).其中format_peername/1emqx_connection_util.erl将 IP 地址与端口格式化为IP:Port形式的二进制串。于是 CONNECT 完成前的关闭报告会得到类似下面的label#{ reason proc_heap_too_large, value 123456, max 33554432, label {tcp:default, 192.168.1.10:52341} }通过监听器名称与对端地址运维人员同样可以反查具体连接。4.3 两种标签的衔接关系从源码结构看两种标签存在明确的阶段分工连接进程创建早期 →emqx_connection_util:label_process/2设置{ListenerId, PeernameStr}CONNECT 完成 →emqx_channel:set_log_meta/2覆盖为ClientId。由于proc_lib:set_label/1是覆盖式写入已建立连接的关闭报告只会携带最新的 Client ID 标签而未完成 CONNECT 的连接则会保留早期的监听器 对端地址标签。add_proc_label/2对两种阶段都一视同仁地读取并注入从而实现了无论连接处于哪个阶段都能识别身份的修复目标。同理其他设置过进程标签的进程如 emqx_logger.erl 中以{clientid, ClientIdSafe}为标签的日志进程也会在关闭报告中获得相应标识。五、修复后的日志形态与排查实战修复生效后connection_shutdown_due_to_oom错误日志中的 Detail 将包含label字段例如#{ msg connection_shutdown_due_to_oom, reason mailbox_overflow, value 1200, max 1000, label client-001 }排查步骤建议以connection_shutdown_due_to_oom为关键字检索节点日志筛选出所有强制关闭事件读取每条记录的label若为二进制字符串即为被关闭客户端的 Client ID可直接在仪表盘或emqx ctl clients list中复核该客户端的历史连接记录注意此时连接已断开列表查询针对的是残留会话信息若label为二元组{ListenerId, PeernameStr}说明连接在握手阶段即被关闭可按监听器名称与对端地址追溯来源结合value与max判断是邮箱溢出mailbox_overflow还是堆过大proc_heap_too_large再反推该客户端的行为特征如是否订阅了高频主题、是否存在异常 PUBLISH 洪泛决定是否上调对应 Zone 的force_shutdown阈值。六、相关测试与验证仓库中与本机制直接相关的测试位于emqx_connection_SUITE.erl覆盖连接层check_oom触发关闭的行为emqx_channel_SUITE.erl 与 emqx_client_SUITE.erl验证客户端连接在邮箱/堆越限场景下的表现emqx_config_SUITE.erl 与 emqx_schema_tests.erl校验force_shutdown各字段的默认值、别名与取值合法性。读者在阅读源码时可重点对照 emqx_utils.erlOOM 判定与 label 注入、emqx_connection.erl触发点与日志输出以及 emqx_connection_util.erl握手期标签设置三个文件即可完整理解本次修复的全貌。小结本次修复变更记录见 fix-18521.en.md通过在关闭原因中注入proc_lib进程标签为label字段解决了强制关闭报告无法识别客户端的运维痛点已建立连接携带 Client ID未完成 CONNECT 的连接携带监听器名称与对端地址。掌握这一机制后运维人员即可在 OOM 类连接中断场景中快速定位问题客户端进而通过调整 Zone 级force_shutdown.max_mailbox_size/force_shutdown.max_heap_size阈值或客户端行为治理来规避大面积断连。赞分享后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载相关推荐EMQX 连接进程 OOM 强制关闭force_shutdown可观测性增强与 WebSocket 网关修复解析EMQX 连接进程 OOM 强制关闭force_shutdown可观测性增强与 WebSocket 网关修复解析 导读 本文聚焦 EMQX 开源仓库 cha后端物联网消息队列通信从理论到实践pull-stream设计目标与实现原理的终极指南从理论到实践pull stream设计目标与实现原理的终极指南 在JavaScript的流处理领域pull stream以其独特的设计理念和极简的实现方式脱后端EMQX 连接进程关闭前的邮箱排空机制修复客户端断连前最后发送的 MQTT 包丢失问题EMQX 连接进程关闭前的邮箱排空机制修复客户端断连前最后发送的 MQTT 包丢失问题 导读 本篇文章基于 EMQX 开源仓库中的变更记录 fix 17172后端物联网消息队列通信上一篇如何用FuxiCTR快速搭建CTR预测模型5分钟上手教程下一篇如何用Barcode Buddy实现智能条码管理3分钟快速上手指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 EMQX 连接强制关闭（force_shutdown）报告的可识别性修复：为 OOM 关闭事件补充客户端标识（label）字段
后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载导读当 EMQX 中某个客户端连接进程的邮箱队列或堆内存超过force_shutdown.max_mailbox_size/force_shutdown.max_heap_size阈值而被强制关闭时旧版本的关闭报告里只有触发的阈值与实测值运维人员无法得知被关闭的究竟是哪个客户端。本篇文章围绕该修复对应仓库变更记录 fix-18521.en.md完整梳理 EMQX 强制关闭机制的配置项、OOM 判定调用链以及本次为关闭报告新增label字段的底层实现原理帮助读者掌握如何定位与排查此类连接被强制关闭事件。一、问题背景强制关闭报告为何难以排查EMQX 为每个客户端连接维护一个独立的 Erlang 进程。为防止单个失控进程拖垮整个节点EMQX 提供了基于邮箱队列长度与进程堆大小的强制关闭force shutdown机制一旦连接进程的message_queue_len或total_heap_size超过配置阈值该连接会被立即终止以换取系统整体稳定性。在修复之前此类关闭产生的报告只包含触发关闭的原因如mailbox_overflow、proc_heap_too_large、实测值value与阈值max却缺少被关闭进程的身份信息。当多个客户端同时连接、批量触发强制关闭时运维人员无法从日志中分辨出具体是哪个客户端被关闭排障往往需要逐条对照进程 ID 与客户端映射效率极低。本次修复的核心改动见 emqx_utils.erl是在关闭原因中追加一个label字段——对于已完成 CONNECT 的已建立连接label保存的是客户端 IDClient ID对于在 CONNECT 完成之前就被关闭的连接label保存的是监听器名称与对端地址peer address。这样无论是已上线客户端还是握手阶段的新连接都能在关闭报告中得到有效辨识。二、force_shutdown 配置项详解force_shutdown属于监听器/Zone 级别的配置。其字段定义位于 emqx_schema.erl配置项类型默认值说明force_shutdown.enableboolean()true是否启用强制关闭机制标记为IMPORTANCE_NO_DOC即不在默认文档中展示的辅助开关force_shutdown.max_mailbox_sizerange(0, inf)1000连接进程邮箱队列message queue的最大长度超过该长度即触发关闭兼容旧别名max_message_queue_lenforce_shutdown.max_heap_sizewordsize()即字大小如32MB32MB连接进程堆的最大大小超过即触发关闭配置了独立的校验器validate_heap_size/1对应的运行时类型定义在 emqx_types.erl 中称为 OOM 策略oom_policy()-type oom_policy() :: #{ max_mailbox_size non_neg_integer(), max_heap_size non_neg_integer(), enable boolean() }.在连接进程启动时Zone 的force_shutdown配置会被读取并用于调整进程堆大小上限见 emqx_connection.erlShutdownPolicy emqx_config:get_zone_conf(Zone, [force_shutdown]), emqx_utils:tune_heap_size(ShutdownPolicy),其中tune_heap_size/1的实现位于 emqx_utils.erl当enable为false或max_heap_size为0时忽略否则将max_heap_size放大 1.5 倍并写入进程标志max_heap_sizekill true, error_logger true使 Erlang VM 在进程堆溢出时以更强的方式介入。注意这里 1.5 倍溢出窗口是一个运行期保护垫实际对外暴露的配置阈值仍是原始max_heap_size。典型配置示例emqx.conf/ Zone 配置zone.default { force_shutdown { enable true max_mailbox_size 1000 max_heap_size 32MB } }提示max_mailbox_size的语义是邮箱队列长度条数并非字节数max_heap_size以字为单位在 64 位平台上每字 8 字节。三、OOM 判定与关闭报告的生成链路连接进程在每次处理完一批入站消息后都会调用check_oom/3检查自身邮箱与堆是否越限见 emqx_connection.erlcheck_oom(Pubs, Bytes, #state{conf #conf{force_shutdown ShutdownPolicy}}) - case emqx_utils:check_oom(ShutdownPolicy) of {shutdown, #{reason : OomReason} Detail} - ?SLOG(error, Detail#{msg connection_shutdown_due_to_oom}), ?tp_ignore_side_effects_in_prod(check_oom_shutdown, #{ policy ShutdownPolicy, incoming_pubs Pubs, incoming_bytes Bytes, shutdown Detail }), %% triggers terminate/2 callback immediately erlang:exit({shutdown, OomReason}); Result - ?tp(debug, check_oom_ok, #{...}) end.关键调用链如下emqx_utils:check_oom(Policy)读取当前进程的message_queue_len与total_heap_size见 emqx_utils.erl并依次比对两个阈值邮箱长度超限 → 原因mailbox_overflow堆大小超限 → 原因proc_heap_too_large。判定逻辑do_check_oom/2只有在0 Max Val阈值有效且实测值严格大于阈值时才判定越限返回{shutdown, #{reason Reason, value Val, max Max}}。连接进程随即通过erlang:exit({shutdown, OomReason})触发terminate/2回调连接被强制关闭同时输出connection_shutdown_due_to_oom错误日志。修复前的关闭原因形如#{ reason mailbox_overflow, value 1200, max 1000 }此时日志只能看到邮箱长度 1200 超过 1000却不知道是谁的邮箱。四、label 字段的注入机制修复的核心实现本次修复在关闭原因中加入了label字段注入点位于 emqx_utils.erl 的add_proc_label/2%% Add the process label (e.g. {clientid, ClientId}) to the shutdown %% reason so the supervisor report identifies the offending process. add_proc_label(_Pid, ok) - ok; add_proc_label(Pid, {shutdown, Reason}) - case proc_lib:get_label(Pid) of undefined - {shutdown, Reason}; Label - {shutdown, Reason#{label Label}} end.它的工作原理是读取 Erlang OTP 的proc_lib进程标签process label如果该进程已被设置过标签就把标签并入关闭原因如果没有标签则保持原样返回。check_oom/2在完成越限判定后立即调用add_proc_label(Pid, Result)emqx_utils.erl因此只要进程设置了标签关闭报告就一定携带label。那么label的值从哪里来EMQX 在不同连接阶段设置了不同的进程标签4.1 已建立连接label 为客户端 ID当 CONNECT 报文处理完成、连接正式建立时通道层会调用 emqx_channel.erl 的set_log_meta/2set_log_meta(_ConnPkt, #channel{clientinfo #{clientid : ClientId} ClientInfo}) - proc_lib:set_label(ClientId), Username maps:get(username, ClientInfo, undefined), ... emqx_logger:set_proc_metadata([{username, Username}, {tns, Tns}]).这里直接以客户端 ID 作为进程标签。因此凡是完成 CONNECT 的连接一旦后续触发强制关闭报告中的label即为该连接的Client ID运维人员可以直接在日志中按客户端 ID 检索。4.2 CONNECT 完成前的连接label 为监听器与对端地址对于尚未完成 CONNECT 就触发关闭的连接例如握手阶段堆积了过量的报文此时还没有可用的 Client ID。EMQX 在连接进程启动早期即设置另一个标签见 emqx_connection_util.erllabel_process( {Type, ListenerName}, #{peername : Peername} _ConnInfo ) - PeernameStr format_peername(Peername), ListenerId emqx_listeners:listener_id(Type, ListenerName), proc_lib:set_label({ListenerId, PeernameStr}), emqx_logger:set_proc_metadata(#{ listener ListenerId, peername PeernameStr }).其中format_peername/1emqx_connection_util.erl将 IP 地址与端口格式化为IP:Port形式的二进制串。于是 CONNECT 完成前的关闭报告会得到类似下面的label#{ reason proc_heap_too_large, value 123456, max 33554432, label {tcp:default, 192.168.1.10:52341} }通过监听器名称与对端地址运维人员同样可以反查具体连接。4.3 两种标签的衔接关系从源码结构看两种标签存在明确的阶段分工连接进程创建早期 →emqx_connection_util:label_process/2设置{ListenerId, PeernameStr}CONNECT 完成 →emqx_channel:set_log_meta/2覆盖为ClientId。由于proc_lib:set_label/1是覆盖式写入已建立连接的关闭报告只会携带最新的 Client ID 标签而未完成 CONNECT 的连接则会保留早期的监听器 对端地址标签。add_proc_label/2对两种阶段都一视同仁地读取并注入从而实现了无论连接处于哪个阶段都能识别身份的修复目标。同理其他设置过进程标签的进程如 emqx_logger.erl 中以{clientid, ClientIdSafe}为标签的日志进程也会在关闭报告中获得相应标识。五、修复后的日志形态与排查实战修复生效后connection_shutdown_due_to_oom错误日志中的 Detail 将包含label字段例如#{ msg connection_shutdown_due_to_oom, reason mailbox_overflow, value 1200, max 1000, label client-001 }排查步骤建议以connection_shutdown_due_to_oom为关键字检索节点日志筛选出所有强制关闭事件读取每条记录的label若为二进制字符串即为被关闭客户端的 Client ID可直接在仪表盘或emqx ctl clients list中复核该客户端的历史连接记录注意此时连接已断开列表查询针对的是残留会话信息若label为二元组{ListenerId, PeernameStr}说明连接在握手阶段即被关闭可按监听器名称与对端地址追溯来源结合value与max判断是邮箱溢出mailbox_overflow还是堆过大proc_heap_too_large再反推该客户端的行为特征如是否订阅了高频主题、是否存在异常 PUBLISH 洪泛决定是否上调对应 Zone 的force_shutdown阈值。六、相关测试与验证仓库中与本机制直接相关的测试位于emqx_connection_SUITE.erl覆盖连接层check_oom触发关闭的行为emqx_channel_SUITE.erl 与 emqx_client_SUITE.erl验证客户端连接在邮箱/堆越限场景下的表现emqx_config_SUITE.erl 与 emqx_schema_tests.erl校验force_shutdown各字段的默认值、别名与取值合法性。读者在阅读源码时可重点对照 emqx_utils.erlOOM 判定与 label 注入、emqx_connection.erl触发点与日志输出以及 emqx_connection_util.erl握手期标签设置三个文件即可完整理解本次修复的全貌。小结本次修复变更记录见 fix-18521.en.md通过在关闭原因中注入proc_lib进程标签为label字段解决了强制关闭报告无法识别客户端的运维痛点已建立连接携带 Client ID未完成 CONNECT 的连接携带监听器名称与对端地址。掌握这一机制后运维人员即可在 OOM 类连接中断场景中快速定位问题客户端进而通过调整 Zone 级force_shutdown.max_mailbox_size/force_shutdown.max_heap_size阈值或客户端行为治理来规避大面积断连。赞分享后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载相关推荐EMQX 连接进程 OOM 强制关闭force_shutdown可观测性增强与 WebSocket 网关修复解析EMQX 连接进程 OOM 强制关闭force_shutdown可观测性增强与 WebSocket 网关修复解析 导读 本文聚焦 EMQX 开源仓库 cha后端物联网消息队列通信从理论到实践pull-stream设计目标与实现原理的终极指南从理论到实践pull stream设计目标与实现原理的终极指南 在JavaScript的流处理领域pull stream以其独特的设计理念和极简的实现方式脱后端EMQX 连接进程关闭前的邮箱排空机制修复客户端断连前最后发送的 MQTT 包丢失问题EMQX 连接进程关闭前的邮箱排空机制修复客户端断连前最后发送的 MQTT 包丢失问题 导读 本篇文章基于 EMQX 开源仓库中的变更记录 fix 17172后端物联网消息队列通信上一篇如何用FuxiCTR快速搭建CTR预测模型5分钟上手教程下一篇如何用Barcode Buddy实现智能条码管理3分钟快速上手指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考