EMQX 大规模连接集群下的会话诊断利器emqx_session_tool 使用与原理全解【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqxEMQX 在承载数万甚至数十万并发连接的集群中运维人员经常需要回答一个朴素却棘手的问题到底哪些客户端积压了消息、哪些在丢消息 逐个翻页浏览客户端列表不现实为此 EMQX 在 6.x 中新增了面向运维的诊断模块emqx_session_tool可从远端控制台直接调用按任意会话指标如mqueue_len、mqueue_dropped、inflight_cnt找出 Top-K 会话。本文基于当前仓库源码完整讲解该模块的 API、指标清单、参数语义、集群聚合机制与底层安全设计让你在线上集群中快速定位问题会话。一、为什么需要 emqx_session_toolEMQX 的会话信息保存在每节点的 channel registryemqx_channel_infoETS 表见 emqx_cm.hrl中。当节点上有数万连接时手动分页遍历客户端列表来寻找积压会话操作成本高、效率低常规的管理接口往往需要拉取全量数据再在客户端侧排序内存与带宽开销大直接向连接进程发消息查询状态会给承载业务流的进程增加额外负载。emqx_session_tool正是为解决这一痛点而生的它在节点本地流式扫描channel registry只保留一个有界的 Top-K 有序集合并且读取的是 registry 中已缓存的会话指标快照不向任何连接进程发消息因此对线上集群足够安全详见 emqx_session_tool.erl 的模块文档。二、快速上手从远端控制台定位 Top-K 会话emqx_session_tool的设计目标是可从远端控制台remote console在运行的集群上直接调用。最简用法%% 找出本地节点上消息队列最长的 20 个会话 emqx_session_tool:top_by(mqueue_len). %% 找出丢消息最多的会话 emqx_session_tool:top_by(mqueue_dropped). %% 找出飞行窗口inflight占用最高的会话 emqx_session_tool:top_by(inflight_cnt).top_by/1等价于top_by(Metric, #{})即使用全部默认参数。返回结果按指标值从高到低排序每行是一个 map包含clientid、pid、node、metric、value等字段。如果集群有多个节点使用cluster_top_by/1一次性聚合所有运行节点的结果%% 在整个集群范围内找出 mqueue_len 最大的 Top-20 会话 emqx_session_tool:cluster_top_by(mqueue_len).每个节点的扫描只遍历本节点自身的会话集合不存在跨节点 ETS 遍历各节点的 Top-K 集合在发起节点合并后重新裁剪为全局 Top-K扫描失败的节点会被跳过其结果行不出现详见 emqx_session_tool.erl。三、可排序的指标metric全清单top_by/1、scan/1等函数接受的指标可通过emqx_session_tool:available_metrics/0实时查询它由两类指标拼接而成见 emqx_session_tool.erl。第一类会话级 gauge?SESSION_STATS_KEYS与 emqx_session_mem 中维护的会话统计对应指标含义subscriptions_cnt/subscriptions_max当前订阅数 / 订阅数上限inflight_cnt/inflight_max当前飞行中消息数 / 飞行窗口上限mqueue_len/mqueue_max当前消息队列长度 / 队列上限mqueue_dropped因队列满等原因丢弃的消息数累计total_payload_bytes队列中累积的载荷总字节数awaiting_rel_cnt/awaiting_rel_max等待 PUBREL 确认的消息数 / 上限注意会话统计中的durable布尔量与next_pkt_id回绕的报文 ID 计数器被有意排除——对它们排序没有实际意义见 emqx_session_tool.erl 的注释。第二类channel 报文/消息计数器?CHANNEL_METRICS定义于 emqx_channel.hrlrecv_pkt、recv_msg、recv_msg.qos0、recv_msg.qos1、recv_msg.qos2、recv_msg.dropped、recv_msg.dropped.await_pubrel_timeoutsend_pkt、send_msg、send_msg.qos0、send_msg.qos1、send_msg.qos2、send_msg.dropped、send_msg.dropped.expired、send_msg.dropped.queue_full、send_msg.dropped.too_large这些计数器可用于回答哪些客户端发送/接收消息最多、哪些消息被丢弃过期/队列满/过大等问题。传入不在清单内的指标会直接报错error({unsupported_metric, Metric, available_metrics()})scan/1未提供metric选项则报error({missing_required_option, metric})见 emqx_session_tool.erl。四、scan_opts 选项详解top_by/2、scan/1、cluster_top_by/2都接受同一个选项 map类型为scan_opts()完整定义见 emqx_session_tool.erl选项默认值含义metric必填用于排序的会话指标必须是available_metrics/0之一top_k20返回的行数上限即最终保留的 Top-K 规模min_value1排除指标值低于该阈值的会话默认 1 意味着默认过滤掉值为 0 的会话chunk1000每次ets:select批处理的行数sleep_ms50每处理完一批后休眠的毫秒数yield 给其他进程extra_keys[]附加到结果行的缓存信息字段从缓存的 session/clientinfo/conninfo 中解析例如created_at、username、peername、connected_at、proto_verextra_stats[]附加到结果行的缓存统计字段例如mqueue_len、total_payload_bytes、inflight_cntrpc_timeout30000cluster_top_by/2时每节点的 RPC 超时毫秒单节点scan/1忽略此项一个包含丰富上下文信息的调用示例emqx_session_tool:top_by(mqueue_len, #{ top_k 10, min_value 100, %% 只关心积压超过 100 条的会话 chunk 500, %% 每批 500 行 sleep_ms 10, %% 批间休眠 10ms降低对调度的影响 extra_keys [username, peername, connected_at], extra_stats [mqueue_len, total_payload_bytes, inflight_cnt] }).结果行结构row()见 emqx_session_tool.erl#{ clientid : binary(), %% 客户端 ID pid : pid(), %% channel 进程 PID node : node(), %% 会话所在节点 metric : atom(), %% 本次排序所用指标 value : number(), %% 该会话的指标值已还原为正数 extras #{atom() term()} %% 仅当指定了 extra_keys / extra_stats 时存在 }几点细节排序键设计为数值大者优先数值相同时按 clientid 升序因此并列时会得到确定性的输出测试用例t_top_k_tie_breaks_by_row_key对此有专门验证extra_stats保留的是参与排序时的那个快照即使扫描结束后会话状态发生变化结果行的extras也维持排名时读到的一致快照见 emqx_session_tool.erlextra_keys在每次选中 Top-K 胜出者后解析一次best-effort若会话在扫描期间已断开这些字段可能为空。五、底层原理流式扫描 有界 Top-Kscan/1的实现路径是scan_acc_new/1构造累加器 →scan_to_end循环推进 →scan_acc_rows/1收尾见 emqx_session_tool.erl。其核心安全保证可以总结为四条流式扫描绝不构建全表列表通过emqx_utils_stream:ets/1以chunk大小的批次对emqx_channel_infoETS 表执行ets:select每批只投影出{ClientId, ChanPid, Stats}三元组较大的 info map 留在表内、只在 K 个胜出者身上解析见 emqx_session_tool.erl内存有界无论会话总数多少中间只维护一个容量为top_k的gb_sets有序集合heap_offer/5只在优于最差元素时才插入输出规模恒定批次间让出调度每处理完chunk行休眠sleep_ms毫秒在scan_to_end的让步点还主动erlang:garbage_collect()回收刚处理完的批次使长扫描保持平坦的堆见 emqx_session_tool.erl不打扰连接进程指标直接读自 registry 中缓存的 stats proplistproplists:get_value(Metric, Stats, 0)全程不给 channel 进程发消息。数据新鲜度与覆盖范围务必知晓新鲜度读到的是连接进程最近一次发布的 stats 快照。连接进程按所在 zone 的 stats 定时器即 idle timeout刷新缓存因此指标值最多可能滞后该间隔若某 zone 的stats.enable为false其连接只在注册时发布一次快照此时 gauge 反映的是连接建立时的状态见 emqx_session_tool.erl。范围仅覆盖注册在本地 channel registry 中的emqx_session_mem会话状态存放在 DSdurable storage中的持久化会话暂未覆盖源码以 TODO 标注计划在支持后为行打上engine mem | persistent_ds标记见 emqx_session_tool.erl。六、增量扫描引擎把扫描搬进 gen_server除了一次跑完的scan/1模块还导出了一组增量扫描 API便于让 gen_server 之类的事件驱动进程自己持有游标、逐批推进scan_acc_new(Opts)构造初始累加器同scan/1的选项sleep_ms由驱动方负责scan_acc(Acc)推进一个chunk批次返回{continue, Acc}或{done, Acc}scan_acc_rows(Acc)把当前 Top-K 有序集合转成结果行可在扫描中途调用——提前中止也能拿到迄今最好的部分结果这就是中止安全的体现。对应的注释与类型定义见 emqx_session_tool.erl。测试用例t_incremental_scan_acc验证了部分推进即可读到部分结果、推进到完成与一次性的scan/1结果完全一致、对已完成累加器再次调用scan_acc/1幂等返回{done, _}见 emqx_session_tool_SUITE.erl。七、集群聚合与上层封装cluster_top_by 的聚合实现cluster_top_by/2的流程见 emqx_session_tool.erl取emqx:running_nodes()作为目标节点集通过emqx_session_tool_proto_v1:scan(Nodes, Opts, Timeout)即erpc:multicall见 emqx_session_tool_proto_v1.erl并行下发扫描仅收集{ok, Rows}的成功结果按value降序整体排序lists:sublist(Sorted, TopK)重新裁剪为全局 Top-K。该 bpapi 在 EMQX6.0.3引入introduced_in() - 6.0.3。scanner / collector供 Dashboard 等上层使用的封装仓库中还有两个配套 gen_server构成更完整的异步 Top 扫描能力由emqx_session_top_proto_v16.3.0引入桥接emqx_session_top_scanner.erl节点本地的增量扫描执行器。start_scan/1接收选项sort、count、batch_size默认 1000、sleep_ms默认 1用emqx_session_tool:scan_acc_new/1scan_acc/1按定时器逐批推进完成或取消时把结果回报给 collector同一时刻只允许一个扫描第二个请求返回{error, {busy, Node}}。注意此处sort选项做了映射mqueue_length - mqueue_len、total_payload_bytes - total_payload_bytes。emqx_session_top_collector.erl集群级收集器run(Opts, CompletionFun)负责在本节点启动扫描并通过erpc:multicall分发到其他节点汇总各节点回报固定附加extra_stats [mqueue_len, total_payload_bytes, inflight_cnt]按sort/count排序后调用完成回调还提供cancel/0尽力而为地向所有节点广播取消与status/0返回running/completed/failed/cancelled/idle状态含cluster_nodes、bad_replies等诊断字段。也就是说emqx_session_tool是底层纯函数式的扫描内核scanner/collector 是其异步化、集群化封装——将来 Dashboard 等管理界面展示Top 会话视图时走的正是这条链路。八、测试佐证行为即契约模块配套的 Common Test 套件 emqx_session_tool_SUITE.erl 从多个维度固化了上述行为可作为理解语义的活文档t_top_by_ranks_by_metric/t_top_k_limits_result按指标降序、受top_k约束t_min_value_filter默认min_value 1会过滤掉值为 0 的会话显式设置阈值可收紧t_extra_keys_populate_extrasextra_keys会按session→clientinfo→conninfo→ 顶层 的顺序在缓存 info map 中查找字段缺失键返回undefinedt_total_payload_bytes_stats_extrasextra_stats只附加到入选 Top-K 的行且与排序快照一致t_unsupported_metric_errors/t_missing_metric_errors非法指标与缺失metric的报错行为t_real_clients_scannable真实连接的客户端通过emqtt建连可经缓存 stats 被扫描到t_cluster_top_by两节点集群中cluster_top_by(mqueue_len, #{top_k 3})正确合并各节点 Top-K且每行标注会话所属节点。九、使用注意事项小结调用位置emqx_session_tool面向远端控制台/诊断场景直接在运行的集群节点上调用即可指标选取先执行emqx_session_tool:available_metrics()确认指标名会话级 gauge 与 channel 计数器是两类不同语义大集群调参连接数很多时可通过增大sleep_ms、调小chunk进一步压低对线上调度的影响RPC 超时rpc_timeout需要给足默认 30 秒因为扫描是边扫描边休眠的自限速过程数据新鲜度结果反映的是连接进程最近一次发布的 stats 快照在stats.enable关闭的 zone 中可能停留在连接建立时刻覆盖范围目前仅覆盖内存会话emqx_session_memDS 持久化会话尚不在扫描范围内只读安全整个扫描过程只读 ETS、不发消息、不修改任何状态适合在生产集群反复执行。凭借流式扫描 有界 Top-K 缓存快照这套设计emqx_session_tool让运维人员在十万级连接的集群上也能以毫秒级开销、恒定内存代价快速回答谁在积压、谁在丢消息这一日常诊断问题。【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqx创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考