后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载导读本文围绕 EMQX 的变更记录 fix-15783 展开系统讲解 MQTT 监听器「连接速率限制」connection rate limit的配置方法、令牌桶限流子系统的内部实现以及本次修复所保证的「配置热更新后限流立即生效」行为。读完本文你将掌握max_conn_rate/max_conn_burst等参数的写法与取值规则理解 zone、listener、channel 三级限流器的协作方式并能从源码层面解释为何旧版本在调高 burst 后有效速率仍显得偏严。一、fix-15783连接速率限制更新未立即生效的修复本次变更记录的原文指出了一个关键问题确保监听器listener更新完成后对连接速率限制的任何更改都立即生效。此前内部限流器状态的一部分并不会直接受配置更改影响。例如调高 burst 速率后实际生效的速率限制可能看起来比预期更严格。其含义可以拆解为三点更新触发点当通过配置热更新emqx:update_config修改监听器的连接速率参数时修改动作本身是成功的但旧版本中底层限流器有一部分内部状态令牌桶的桶状态没有随配置同步刷新表象用户把max_conn_burst从1/1m调高到20/1m后预期立刻能放行更多突发连接但实际上新连接仍被拒绝仿佛限制没有放宽修复目标让所有与连接速率相关的内部限流器状态在监听器更新完成后立即重建/同步使新配置即刻生效。仓库中的测试用例 t_max_conn_rate_update 正是对这一行为回归验证它先以max_conn_rate 1/1s、max_conn_burst 1/1m建立监听器确认速率与突发配额耗尽后连接被拒绝随后通过emqx:update_config([listeners, Type, Name], {update, #{max_conn_burst 20/1m}})在线调高突发配额并断言紧接着发起的 10 个并发连接全部成功——即新配置无需重启、无需等待即可生效。二、连接速率限制的配置方式EMQX 的速率限制既可以配置在 zone 级别也可以配置在监听器级别。与连接数/连接速率直接相关的限流器名为max_conn其配置键遵循「limiter_name_rate/limiter_name_burst」的约定见 emqx_limiter.erl 的config/2说明与 emqx_limiter_schema.erl 中的限流器名称定义因此对应到监听器配置即max_conn_rate连接速率每秒允许新建的连接数max_conn_burst突发配额允许在更长时间窗口内累积的额外连接数max_connections最大并发连接数独立于限流器但仍通过 ranch limiter 与速率限制联动。仓库提供的 rate_limit.conf.example 给出了一个完整的监听器级示例## Rate Limit ## Rate limiting is applied on the listener listeners.tcp.my_tcp_listener_name { ## Connections per second per listener max_conn_rate 1000/s ## Incoming messages per second per client messages_rate 1000/s ## Incoming message size in bytes per second per client bytes_rate 1MB/s }2.1 速率/突发配额的取值语法rate与burst的字符串解析逻辑位于 emqx_limiter_schema.erl 的parse_rate/1支持以下形式形式示例含义infinitymax_conn_rate infinity不限制仅 rate 支持burst 不允许设为 infinity纯数字每秒1000每秒 1000 个令牌数字/时间1000/s、5/1000ms、10/2s、10/h每时间间隔生成对应数量令牌带字节单位1MB/s、10KB/2s、1GB/h、500B/s用于 bytes 类限流器b、kb、mb、gb按 1024 进制换算时间单位支持d、h、m、s、ms且间隔不能为 0rate的容量也不能为 0。rate决定令牌的稳定生成速率burst决定在更长窗口如1/1m表示每 1 分钟补充 1 个突发令牌内额外累积的令牌量用于容忍瞬时突增流量。2.2 三级限流器与配置归属从源码宏定义emqx_limiter.erl可以看出限流器按作用域分组?LISTENER_LIMITS [max_conn]——监听器级每个监听器一份以{listener, ListenerId}为组?ZONE_LIMITS [max_conn, messages, bytes]——zone 级以{zone, Zone}为组监听器与 zone 的max_conn会叠加生效?CHANNEL_LIMITS [messages, bytes, subscribes]、?SESSION_LIMITS [delivery_bytes, delivery_messages]——客户端连接级以{channel, ListenerId}为组。因此连接速率限制实际是复合限流一个新建连接既要通过 zone 的max_conn配额也要通过监听器的max_conn配额两者由emqx_limiter_composite组合判定见create_listener_limiter/3。三、限流子系统架构令牌桶与两种共享模型EMQX 的限流子系统位于 apps/emqx/src/emqx_limiter/其核心概念如下Limiter限流器建模为一个令牌桶通过全局唯一 ID{Group, Name}标识例如{{zone, default}, messages}Client限流客户端与某个限流器连接负责消费令牌、归还令牌由emqx_limiter_client模块实现两种桶模型共享限流器shared连接到同一限流器的所有客户端共享同一个令牌桶协同消费emqx_limiter_shared独占限流器exclusive每个客户端拥有独立令牌桶互不影响emqx_limiter_exclusive。监听器/zone 的max_conn采用共享模型因为连接速率是监听器全局口径而消息/字节等客户端级限制采用独占模型。从 emqx_limiter/README.md 可知令牌消费依赖三类状态桶设置bucket settingsrate、burst 等参数由emqx_limiter_registry缓存在persistent_term中全局只读快照、查询开销极低桶状态bucket state当前令牌数、上次更新时间等。共享限流器以原子值建模保证并发消费的一致性独占限流器的桶状态内嵌在客户端状态中由客户端按桶设置算法自充填客户端状态client state由持有客户端引用的进程如连接进程管理。入口门面模块emqx_limiter提供组的新建/更新/删除、限流客户端获取connect/1基于 persistent_term 的轻量查找以及 zone/监听器配置驱动的回调。3.1 监听器接入限流的位置TCP 类监听器通过 esockd 接入在 esockd_opts/4 中调用emqx_limiter:create_esockd_limiter_client(Zone, ListenerId)生成限流客户端并作为limiter选项注入 esockd。esockd 侧的回调实现见 emqx_esockd_limiter.erl当try_consume失败配额不足时返回{pause, 100, ...}即暂停 100ms再接受下一个连接并输出限流告警日志listener_accept_throttled_due_to_quota_exceeded。QUIC 等基于 ranch 的监听器则在 ranch_limiter_opts/4 中构造emqx_ranch_limiter同时携带rate_limiter连接速率与max_connections最大并发连接数并用num_conns_sups num_acceptors保证限流行为的可预期性。四、配置热更新为何会失效以及修复如何生效4.1 监听器更新路径监听器配置更新入口是 emqx_listeners.erl 的update_listener/4update_listener(Type, Name, OldConf, NewConf) - ListenerId listener_id(Type, Name), case is_running(Type, ListenerId, NewConf) of true - ok emqx_limiter:update_listener_limiters(ListenerId, NewConf), do_update_running_listener(Type, Name, OldConf, NewConf); ...注意这里的顺序先更新限流器再更新运行中的监听器。emqx_limiter:update_listener_limiters/2emqx_limiter.erl会同时重建两组限流器update_listener_limiters(ListenerId, ListenerConfig) - ListenerLimiters listener_limiter_options(ListenerConfig), ClientLimiters client_limiter_options(ListenerConfig), ok update_group(listener_group(ListenerId), ListenerLimiters), ok update_group(channel_group(ListenerId), ClientLimiters).update_group/2emqx_limiter.erl进一步做差异比较只有当新旧限流器参数存在差异时才执行「重新注册组 通知底层模块更新」两步update_group(Group, Options) - case emqx_limiter_registry:find_group(Group) of undefined - error({limiter_group_not_found, Group}); {Module, OldOptions} - Diff lists:foldl(fun lists:delete/2, OldOptions, Options), Diff / [] andalso begin ok emqx_limiter_registry:register_group(Group, Module, Options), ok Module:update_group(Group, Options) end, ok end.4.2 内部状态未同步是失效根因emqx_limiter_registry将每个组的限流器参数写入 persistent_termemqx_limiter_registry.erl所有限流客户端在connect/1时通过find_group/1从 persistent_term 读取最新桶设置。问题出在桶设置persistent_term更新了但运行中的客户端持有的旧桶状态没有重置——尤其对共享限流器桶状态由原子值建模若只更新了配置而没有让底层模块同步重建桶状态则新连接仍按旧容量/旧突发配额扣减表现为调高 burst 后有效速率依然偏严。fix-15783 的修复正是补齐了这条同步链使update_group触发的模块级更新能立即反映到令牌桶的实际状态上。从修复意图看这也是为何测试用例 t_max_conn_rate_update 特别强调right after更新后立即它用emqx:update_config在线调高max_conn_burst后不做任何等待直接并发建立 10 条连接并断言全部pong以此锁死更新即生效的语义防止回归。4.3 更新之外的创建与删除路径与更新对称监听器的生命周期管理同样调用限流器门面创建start_listener内调用emqx_limiter:create_listener_limiters(ListenerId, Conf)emqx_listeners.erl失败回滚时调用delete_listener_limiters/1停止stop_listener调用emqx_limiter:delete_listener_limiters(Id)emqx_listeners.erl其中try_delete_group对已不存在的组静默容忍zone 变更emqx_limiter:post_zone_config_update/2依据配置 diff 的added / removed / changed三组分别执行创建、删除默认 zone 永不删除与更新emqx_limiter.erl。五、相关限流参数速查在 zone 或监听器配置中可用的限流参数来自 emqx_limiter_schema.erl 与 rate_limit.conf.example参数作用域说明max_conn_rate/max_conn_burstzone / listener连接速率与突发配额fix-15783 直接涉及的参数max_connectionslistener最大并发连接数类型为infinity | pos_integer()默认见emqx_listeners:default_max_conn()emqx_schema.erlmessages_rate/messages_burstzone / listener / channel每客户端消息接收速率bytes_rate/bytes_burstzone / listener / channel每客户端字节接收速率支持 KB/MB/GB 单位delivery_messages_*/delivery_bytes_*session会话投递方向的消息/字节限流subscribes_*channel订阅操作限流配置形式可直接参考仓库中的示例文件rel/config/examples/rate_limit.conf.example监听器级限流示例rel/config/examples/listeners.tcp.conf.examplemax_connections等监听器基础参数六、小结与验证建议fix-15783 修复的是一个典型的配置已更新、运行态未同步问题其价值在于运维可预期性在线调优连接速率参数时不再需要重启监听器或等待限流器自然冷却新配额即刻反映到令牌桶上。如需在本地验证可参照测试套件 emqx_listeners_limits_SUITE.erl 的三个核心用例t_max_conns验证max_connections全局并发上限超限连接被快速拒绝、断开后可再次接入t_max_conn_rate验证max_conn_rate按监听器而非按 acceptor全局计数配额耗尽即拒绝、冷却后恢复t_max_conn_rate_update验证本文核心主题——max_conn_burst热更新后立即生效。这三个用例覆盖了 tcp / ws / wss 三种监听器协议是理解并回归验证连接速率限制即时生效行为最直接的入口。赞分享后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载相关推荐EMQX 连接速率限制热更新原理监听器更新后限流立即生效的机制解析EMQX 连接速率限制热更新原理监听器更新后限流立即生效的机制解析 导读 本文围绕 EMQX 的一条限流修复记录展开 监听器Listener配置更新完成后端物联网消息队列通信EMQX 为 WS/WSS 监听器补齐连接速率限制max_conn_rate、max_conn_burst 生效与 max_connections 行为变更解析EMQX 为 WS/WSS 监听器补齐连接速率限制max_conn_rate、max_conn_burst 生效与 max_connections 行为变更解后端物联网消息队列通信EMQX 监听器连接速率限制恢复 per-listener 语义max_conn_rate / max_conn_burst 配置迁移指南EMQX 监听器连接速率限制恢复 per listener 语义max_conn_rate / max_conn_burst 配置迁移指南 本文面向从 EMQ后端物联网消息队列通信创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考