EMQX Durable Storage 事务配置运行时热更新：原理、配置与源码解析

后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载EMQX 的 Durable Storage持久化存储简称 DS为消息、会话、定时器等数据提供基于 RocksDB 的持久化能力其中乐观事务Optimistic Transaction子系统负责原子地读写数据。本文围绕 EMQX 6.x 中一项关键能力展开Durable Storage 的事务配置可以在运行期直接修改无需重启节点此前修改该配置必须重启节点才能生效。读完本文你将掌握transaction配置项的完整参数语义、通过emqx_conf:update触发热更新的调用链以及如何在生产环境中安全地动态调整事务参数。变更背景从重启生效到运行时生效本仓库的变更记录 changes/ee/fix-16129.en.md 对这一改进给出了明确说明Durable storage transaction configuration can be changed in the runtime. Previously changing this configuration required a node restart.即在旧版本中修改 Durable Storage 的transaction乐观事务配置后需要重启 EMQX 节点才能让新配置生效本次变更后配置修改会在运行时即时下发到对应数据库并生效避免了因调整事务参数而造成的服务中断。这项能力背后是一整套动态配置机制涉及配置 Schema、配置处理器回调、DS 核心 API 三个层面下面逐一展开。Durable Storage 与乐观事务子系统概览在深入配置项之前先建立整体认知。Durable Storage 应用emqx_durable_storage是 EMQX 内部持久化存储的核心应用其设计目标是每条消息在每个副本上只存储一次无论消费者在线或离线参见 apps/emqx_durable_storage/README.md。数据组织层级Database → Shard → Generation → Stream。数据库彼此独立可按clientId或主题分片每个分片再按时间划分为 Generation消息最终组织在 Stream 中按序写入。事务能力emqx_ds:trans/2可用于原子地执行多次读写操作底层由 emqx_ds_optimistic_tx.erl 中的乐观事务 Leader 进程跟踪近期写入、校验冲突在检测到读写冲突时拒绝提交相关事务。内置后端builtin_raftRaft 复制默认与builtin_local本地单副本均基于 RocksDB。乐观事务的写入并非即时落盘而是由 Leader 以批处理方式按flush_interval/idle_flush_interval节奏刷新并通过conflict_window定义冲突检测窗口、max_pending限制积压事务数量。这些参数共同决定事务吞吐、延迟与冲突行为正是本次支持运行时热更新的对象。transaction配置项完整参数说明Durable Storage 的全部配置由 apps/emqx/src/emqx_ds_schema.erl 定义。transaction字段属于common_builtin_fields/1类型为optimistic_transaction即乐观事务配置位于每个数据库配置之下如durable_storage.messages.transaction。从fields(optimistic_transaction)emqx_ds_schema.erl可以看到以下四个可热更新的参数参数类型默认值说明conflict_window时长ms 级5s乐观事务冲突检测窗口窗口内发生的读写冲突会被检测并拒绝提交flush_interval时长ms 级10ms事务批处理的定时刷新间隔idle_flush_interval时长ms 级1ms空闲无新事务积压时的刷新间隔用于降低空闲延迟max_pending正整数10000允许积压的待提交事务数量上限配置示例HOCON 格式durable_storage { messages { transaction { conflict_window 5s flush_interval 10ms idle_flush_interval 1ms max_pending 10000 } } }值得注意的细节不同数据库默认值不同。在schema/0中sessions持久会话与timers持久定时器两个数据库的默认配置显式覆盖了idle_flush_interval 0ms见 emqx_ds_schema.erl因为这类数据库对延迟敏感希望空闲时立即刷新而通用默认值为1ms。映射关系。transaction配置在运行时通过translate_otx_opts/1转换为 DS 内部选项#{conflict_window, flush_interval, idle_flush_interval, max_items}max_pending→max_items再交由后端应用emqx_ds_schema.erl。哪些 DB 受影响。config_root_to_dbs/1emqx_ds_schema.erl定义了配置根与数据库的对应关系messages、sessions、timers、shared_subs为一一对应而 MQ 状态、MQ 消息、Streams 消息等配置根会展开映射到多个数据库如mq_messages同时作用于mq_messages_lastvalue与mq_messages_regular。运行时热更新的完整调用链本次变更的核心在于让配置更新沿着 EMQX 标准动态配置链路直达数据库运行时。调用链如下emqx_conf:update(...) │ ▼ emqx_config_handler配置变更分发 │ ▼ emqx_ds_schema:post_config_update/6 │ durable_storage.DB.transaction 路径 ▼ emqx_ds_schema:update_db_config/2逐 DB 应用 │ ▼ emqx_ds:update_db_config/2DS 核心 API │ ▼ Backend:verify_db_opts Backend:update_db_config后端实际生效各环节的源码证据如下1. Schema 注册与回调入口emqx_ds_schema实现了emqx_config_handler行为-behaviour(emqx_config_handler)见 emqx_ds_schema.erl并通过add_handler/0注册了对durable_storage命名空间下各级路径的处理emqx_ds_schema.erl。2. post_config_update配置落地的分发点当配置变更提交后post_config_update/6根据变更路径分发处理emqx_ds_schema.erlpost_config_update( [durable_storage, db_groups, Group | _], _, _, _, _, _ ) - setup_db_group(Group, translate_db_group(emqx_config:get([durable_storage, db_groups, Group]))); post_config_update([durable_storage, Config | _], _, _, _, _, _) - lists:foreach( fun(DB) - update_db_config(DB, db_config(Config)) end, config_root_to_dbs(Config) );即凡是对durable_storage下数据库配置含transaction的更新都会把该配置根映射到的每一个数据库逐一调用update_db_config/2进行热更新。3. update_db_config静默处理未打开数据库update_db_config/2emqx_ds_schema.erl调用emqx_ds:update_db_config(DB, Conf)并通过?tp打点ds_db_runtime_config_update与ds_db_runtime_config_update_failed事件便于链路追踪与测试断言。当数据库尚未打开时返回{error, recoverable, db_is_closed}此时被当作正常情况忽略ok即配置会在数据库下次打开时生效不会导致更新失败。4. emqx_ds:update_db_config核心 APIDS 核心模块 emqx_ds.erl 中的update_db_config/2是配置真正落地的入口-spec update_db_config(db(), #{atom() _}) - ok | error(_). update_db_config(DB, Patch) - ?with_dsch( DB, #{cbm : Mod}, maybe #{} ? emqx_dsch:get_db_runtime(DB), {ok, NewSchema, NewRTConf} ? Mod:verify_db_opts(DB, Patch), Mod:update_db_config(DB, NewSchema, NewRTConf) else ... end ).从源码可以看出热更新的执行前提数据库必须处于已打开状态emqx_dsch:get_db_runtime(DB)返回空表时返回db_is_closed新配置必须先通过后端回调verify_db_opts/2的校验校验通过后调用后端模块如emqx_ds_builtin_local/emqx_ds_builtin_raft的update_db_config/3将NewSchema与NewRTConf写入数据库运行时状态。同时 emqx_ds.erl 的文档明确指出并非所有配置都支持运行时修改——backend后端类型、n_shards分片数以及某些语义标志semantic flags在数据库创建后不可更改而事务参数这类运行时配置则属于可热更新范围。这与本次变更transaction配置运行时生效保持一致。测试验证动态重配置用例仓库的测试用例直接验证了事务配置可运行时热更新这一能力。在 apps/emqx/test/emqx_persistent_messages_SUITE.erl 中t_runtime_config_update/1的注释写道This testcase verifies dynamic reconfiguration capability ofemqx_ds_schemamodule.该用例遍历所有配置根与依赖数据库的映射关系对每个数据库执行emqx_conf:update( [durable_storage, ConfRoot, transaction, idle_flush_interval], 10, #{persistent false} )即通过emqx_conf:update/3将每个数据库的transaction.idle_flush_interval动态更新为10并用?wait_async_action等待ds_db_runtime_config_update事件出现以断言配置确实在运行时被下发到了对应的数据库。#{persistent false}表示这是一次非持久化的运行时更新仅作用于当前运行中的节点。该用例覆盖的数据库包括timers、messages持久消息、sessions持久会话状态、mq_states、mq_messages_regular、mq_messages_lastvalue、streams_messages_regular、streams_messages_lastvalue证明热更新机制对所有 Durable Storage 数据库统一生效。实际操作如何在运行期调整事务参数结合上述机制在生产环境中动态调整事务配置有两种方式方式一通过emqx_conf:update热更新本次变更的核心能力%% 在 Erlang Shell / 管理通道中执行 emqx_conf:update( [durable_storage, messages, transaction, idle_flush_interval], 2ms, #{persistent true} ).第 3 个参数#{persistent true}表示同时持久化到配置文件重启后依然生效#{persistent false}则仅对当前运行期生效。修改会沿前述调用链即时下发到对应的数据库后端无需重启节点。方式二修改配置文件后由配置热加载机制触发在emqx.conf中调整durable_storage.DB.transaction.*的值并触发配置重载如通过 Dashboard 或emqx_ctl重新加载配置emqx_config_handler同样会走post_config_update/6分发到update_db_config/2实现与方式一等价的运行时生效。注意事项可修改范围transaction下的四个参数conflict_window、flush_interval、idle_flush_interval、max_pending均支持运行时热更新而backend、n_shards等创建期参数不可在数据库创建后修改见 emqx_ds.erl 的说明。持久化策略热更新默认只改运行态如需跨重启保留务必使用persistent true或同步更新配置文件。多数据库联动由于一个配置根可能映射多个数据库如mq_messages同时作用两个库修改这类配置会同时下发到全部关联数据库。总结从改配置必须重启节点到事务配置运行期即时生效本次变更通过打通emqx_conf:update→emqx_ds_schema:post_config_update→emqx_ds:update_db_config→ 后端verify_db_opts/update_db_config的完整链路将 Durable Storage 乐观事务的conflict_window、flush_interval、idle_flush_interval、max_pending四个参数纳入动态配置体系。运维人员现在可以在不中断服务的前提下根据业务流量实时调优事务批处理节奏与冲突窗口并由 emqx_persistent_messages_SUITE.erl 等测试用例持续保障该能力的正确性。相关源码可进一步参阅 emqx_ds_schema.erl、emqx_ds.erl 与 emqx_ds_optimistic_tx.erl。赞分享后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载相关推荐Apollo配置实时更新原理深入理解配置热发布机制Apollo配置实时更新原理深入理解配置热发布机制 Apollo作为携程开源的分布式配置管理平台其 配置热发布机制 是实现实时更新的核心技术。通过巧妙的设计配置中心后端微服务5个必知的PowerShellEditorServices高级功能从代码导航到实时语义分析5个必知的PowerShellEditorServices高级功能从代码导航到实时语义分析 PowerShellEditorServices是微软开发的Pow开发工具iii 运行时配置指南configuration worker 的双层配置体系与热更新实战iii 运行时配置指南configuration worker 的双层配置体系与热更新实战 iii当前仓库将运行时配置拆分为两层启动时由引擎读取的 co后端流程编排任务调度可观测性创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考