EMQX 监听器连接速率限制配置与热更新即时生效机制解析-平芜编程栈

后端
物联网
消息队列
通信

【免费下载链接】emqx

The 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，支持以下形式：

形式	示例	含义
`infinity`	`max_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，且间隔不能为 0；rate的容量也不能为 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 settings）：rate、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/4：

update_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/2（emqx_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/2（emqx_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_term（emqx_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_burst`	zone / listener	连接速率与突发配额（fix-15783 直接涉及的参数）
`max_connections`	listener	最大并发连接数，类型为`infinity \| pos_integer()`，默认见`emqx_listeners:default_max_conn()`（emqx_schema.erl）
`messages_rate`/`messages_burst`	zone / listener / channel	每客户端消息接收速率
`bytes_rate`/`bytes_burst`	zone / listener / channel	每客户端字节接收速率（支持 KB/MB/GB 单位）
`delivery_messages_`/`delivery_bytes_`	session	会话投递方向的消息/字节限流
`subscribes_*`	channel	订阅操作限流