后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载导读本文基于 EMQX 仓库的变更记录 fix-15706.en.md修复项编号 #15706深入剖析一个与消息处理链路数据一致性相关的缺陷当用户删除一条 Message Transformation消息转换或 Schema ValidationSchema 校验后再禁用排在它后面的某一条规则被禁用的规则却可能仍然生效。文章将结合 emqx_message_transformation 与 emqx_schema_validation 两个应用的源码实现、测试用例和 HTTP API说明该问题的根因、修复方式以及如何通过测试与 API 复现和验证。一、问题背景删除后禁用规则却仍然生效官方变更记录原文fix-15706.en.md描述的缺陷如下Fixed an issue that could make Message Transformations and Schema Validations behave inconsistently. If one deleted a Message Transformation / Schema Validation, and then disabled one of the items that followed the deleted one, it would remain enabled.同一版本发布说明 e5.8.8.en.md 中给出了更具体的描述Fixed an indexing issue that could cause Message Transformations and Schema Validations to behave inconsistently. Deleting one item could corrupt the topic index, so that a subsequent item remained active even after being disabled.综合两条记录可以确认问题本质影响范围Message Transformations消息转换与 Schema ValidationsSchema 校验两类功能同时受影响因为它们共享几乎相同的底层索引机制触发序列先删除列表中的某一条规则再禁用紧跟在它之后的某一条规则错误表现被禁用的规则仍然匹配并作用于消息发布流程表现为禁用不生效。问题的根因指向topic index主题索引被删除操作破坏删除规则后后续规则的索引位置没有被正确重建导致禁用操作写入的索引信息与配置顺序错位。二、底层机制为何删除会污染禁用要理解这个 Bug需要先了解这两类功能在 EMQX 中是如何被存储与匹配的。2.1 双 ETS 表规则表 主题索引表在 emqx_message_transformation_registry.erl 中定义了两张 ETS 表-define(TRANSFORMATION_TOPIC_INDEX, emqx_message_transformation_index). -define(TRANSFORMATION_TAB, emqx_message_transformation_tab).规则表emqx_message_transformation_tab以规则名Name为 key 存储完整的规则定义enable、topics、operations、payload_decoder等主题索引表emqx_message_transformation_index将规则配置的主题与规则的位置索引Pos关联起来用于在消息发布时快速匹配。Schema Validation 侧完全镜像了这套设计见 emqx_schema_validation_registry.erl-define(VALIDATION_TOPIC_INDEX, emqx_schema_validation_index). -define(VALIDATION_TAB, emqx_schema_validation_tab).2.2 位置索引Pos顺序与启用的核心EMQX 的配置以有序列表维护规则每一条规则在列表中的序号即其位置索引Pos。在 do_insert 中可以看到索引写入逻辑do_insert(Pos, Transformation) - #{ enable : Enabled, name : Name, topics : Topics } Transformation, maybe_create_metrics(Name), do_insert_into_tab(Name, Transformation, Pos), Enabled andalso upsert_topic_index(Name, Pos, Topics), ok.关键点在Enabled andalso upsert_topic_index(Name, Pos, Topics)只有enable true的规则才会把主题写入索引表。禁用规则时enable false对应主题会从索引中删除启用时再写回。当消息发布时on_message_publish 通过emqx_message_transformation_registry:matching_transformations(Topic)查询索引do_matching_transformations 会先emqx_topic_index:matches/3拿到匹配条目再按Pos排序后返回规则Transformations1 lists:keysort(1, Transformations0),也就是说主题索引表里残留的任何条目都会让规则被匹配并执行这与配置中的enable字段无关——索引表才是运行时生效的唯一依据。2.3 修复前的缺陷链路修复前的删除逻辑do_delete只做了三件事do_delete(Transformation, Pos) - #{ name : Name, topics : Topics } Transformation, ets:delete(?TRANSFORMATION_TAB, Name), delete_topic_index(Pos, Topics), drop_metrics(Name), ok.它按删除那一刻的Pos清理了索引。但删除之后后面所有规则的Pos在配置层面都前移了一位索引表却仍然保留着它们旧的Pos。此时用户禁用删除后变成第 1 条的那条规则禁用操作按新位置Pos 1去索引表删除条目但索引表中该规则的主题记录挂在旧位置Pos 2甚至与别的规则位置混淆之下结果旧的索引条目成为孤儿数据规则在运行时仍然被matching_transformations/1命中表现为禁用了却依然生效。由于发布说明中明确指出是Deleting one item could corrupt the topic index这与上述从代码结构推断出的位置错位完全吻合。三、修复方案删除后强制重建位置索引修复的核心在于删除规则后不再只依赖按旧Pos清理而是对整张索引表做一次完整的位置重排。3.1 配置层delete 与 reorder 统一走 reindex在 emqx_message_transformation_config.erl 的post_config_update中删除分支现在多了一次重排调用post_config_update(?TRANSFORMATIONS_CONF_PATH, {delete, Name}, New, Old, _AppEnvs) - {Pos, Transformation} fetch_with_index(Old, Name), ok emqx_message_transformation_registry:delete(Transformation, Pos), ok emqx_message_transformation_registry:reindex_positions(New, Old), ok; post_config_update(?TRANSFORMATIONS_CONF_PATH, {reorder, _Order}, New, Old, _AppEnvs) - ok emqx_message_transformation_registry:reindex_positions(New, Old), ok;删除delete与重排reorder现在都调用reindex_positions/2以新配置列表New与旧配置列表Old为输入让索引表与配置顺序严格对齐。Schema Validation 侧在 emqx_schema_validation_config.erl 有完全相同的修复。3.2 索引层先清空旧索引再按新顺序重建reindex_positions/2的实现见 emqx_message_transformation_registry.erldo_reindex_positions(NewTransformations, OldTransformations) - lists:foreach( fun({Pos, Transformation}) - #{topics : Topics} Transformation, delete_topic_index(Pos, Topics) end, lists:enumerate(OldTransformations) ), lists:foreach( fun({Pos, Transformation}) - #{ name : Name, topics : Topics } Transformation, do_insert_into_tab(Name, Transformation, Pos), upsert_topic_index(Name, Pos, Topics) end, lists:enumerate(NewTransformations) ).该函数分两步执行清理阶段对旧列表按lists:enumerate/1重新编号逐个删除旧索引条目——这一步保证了删除后遗留的孤儿索引位于错误Pos的条目被彻底清除重建阶段对新列表按新顺序重新编号逐个把规则写回规则表并重建主题索引。lists:enumerate/1从 1 开始连续编号恰好与配置序号的语义一致确保索引中的Pos↔ 配置中的序号始终一一对应。修复后禁用操作写入索引的Pos必然与索引表中实际存储的Pos一致disable才能真正生效。值得注意的还有do_update/3emqx_message_transformation_registry.erl更新规则时先delete_topic_index(Pos, OldTopics)再按需upsert_topic_index它天然按新Pos处理因此不受该缺陷影响——问题只在删除路径上。四、测试用例如何复现与验证该缺陷修复配套的回归测试位于 emqx_message_transformation_http_api_SUITE.erl测试名t_delete_then_disable直接对应删除后禁用这一复现序列%% Original issue: https://emqx.atlassian.net/browse/EMQX-14604 t_delete_then_disable(_TCConfig) - %% 1) Create two enabled MTs/SVs (no need to have overlapping topics). Name1 will_be_deleted, Transformation1 transformation(Name1, [dummy_operation()]), {201, _} insert(Transformation1), Name2 will_be_disabled, Transformation2 transformation(Name2, [dummy_operation()], #{ topics [u/] }), {201, _} insert(Transformation2), %% 2) Delete MT/SV no. 1. This will make no. 2 become no. 1. {204, _} delete(Name1), %% 3) Disable the new no. 1 (previous no. 2). {204, _} disable(Name2), %% 4) Original problem: metrics keep increasing (topic index is corrupt and still %% triggers the MT/SV). Topic u/1, ?assertIndexOrder([], Topic), C connect(c1), {ok, _, [_]} emqtt:subscribe(C, u//), ok publish(C, Topic, #{t t}), ?assertNotReceive({publish, _}), ok.测试完整还原了缺陷场景依次创建两条启用状态的规则第二条主题为u/删除第一条规则——按缺陷描述此时第二条从位置 2变成位置 1但索引表里仍残留旧位置条目禁用现已成为第 1 条的第二条规则向u/1发布消息断言?assertIndexOrder([], Topic)表示该主题下不应匹配到任何规则且客户端不应收到任何消息?assertNotReceive({publish, _})。如果索引未正确重建第 4 步会失败被禁用的规则仍会拦截/转换消息matched等指标持续增长测试注释中明确指出metrics keep increasing是原始问题的外在表现。该测试同样覆盖了 Schema Validation——测试注释写明 Create two enabled MTs/SVs即两个模块共用同一套复现逻辑。五、复现与验证的操作路径5.1 相关 HTTP API两类功能均通过数据集成?SCOPE_DATA_INTEGRATION相关的 HTTP API 管理见 emqx_message_transformation_http_api.erl/message_transformations GET 列出 / POST 追加 / PUT 更新 /message_transformations/reorder POST 重排顺序 /message_transformations/dryrun POST 试运行 /message_transformations/transformation/:name GET 查询 / DELETE 删除 /message_transformations/transformation/:name/enable/:enable PUT 启用或禁用 /message_transformations/transformation/:name/metrics GET 指标 /message_transformations/transformation/:name/metrics/reset POST 重置指标对应地Schema Validation 的 API 基路径为/schema_validations见 emqx_schema_validation_http_api.erl。5.2 手动复现步骤通过POST /message_transformations依次创建两条规则第二条的主题设为u/并保持enable: true通过DELETE /message_transformations/transformation/will_be_deleted删除第一条规则通过PUT /message_transformations/transformation/will_be_disabled/enable/false禁用第二条规则向主题u/1发布一条消息并订阅验证修复后消息应原样透传、不被转换且GET /message_transformations/transformation/will_be_disabled/metrics中matched指标不再增长。5.3 运行回归测试仓库内可直接运行该测试套件需先按 CONTRIBUTING.md 准备构建环境例如执行消息转换应用的 HTTP API 测试套件make apps/emqx_message_transformation/ct # 或单独运行该用例 ./scripts/ct/run.sh apps/emqx_message_transformation -suite emqx_message_transformation_http_api_SUITE -case t_delete_then_disableSchema Validation 侧对应测试套件为 emqx_schema_validation_http_api_SUITE.erl。六、修复的工程价值与启示本次修复虽然只涉及一个变更记录但揭示了 EMQX 规则类功能一个值得复用的设计原则索引表是运行时生效的唯一依据enable只是配置层面的标志真正驱动message.publish钩子emqx_message_transformation.erl 中的register_hooks/0行为的是主题索引表。任何修改配置顺序、增删规则的操作都必须保证索引表与配置严格同步删除 ≠ 只删自己删除有序结构中的元素会引发后续元素的位置漂移因此删除路径必须像重排路径一样触发全量重建reindex_positions/2而不能只按旧位置做局部清理测试要复现用户操作序列t_delete_then_disable的价值在于它模拟了真实用户先删后禁的组合操作而非孤立的删除或禁用单测——这类操作序列级的回归测试最能捕获索引类状态不一致缺陷。此外从代码结构看Message Transformations 与 Schema Validations 共用同一套规则表 主题索引表 位置重排架构对比两个 registry 模块的do_insert/2、do_delete/2、do_reindex_positions/2几乎逐行对应因此本次修复在 emqx_message_transformation_config.erl 与 emqx_schema_validation_config.erl 两处同步落地确保两个功能的行为保持一致。七、小结fix-15706 修复的是一个典型的有序配置 外部索引场景下的位置错位缺陷删除规则后未重建主题索引导致后续规则的旧索引条目残留进而使禁用操作失效。修复通过在删除与重排路径统一调用reindex_positions/2将索引表与配置顺序重新对齐并用t_delete_then_disable回归测试锁定了该行为。对于在 EMQX 上大量使用消息转换与 Schema 校验的用户而言升级到包含该修复的版本对应 e5.8.8.en.md 发布说明后可以放心执行删除 禁用的组合操作无需再担心规则意外残留生效。赞分享后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载相关推荐EMQX Durable Storage 一致性修复与订阅迭代器边界修复技术解析EMQX Durable Storage 一致性修复与订阅迭代器边界修复技术解析 导读 本文聚焦 EMQX 开源仓库变更记录 changes/ee/fix 16后端物联网消息队列通信Stripe Go 多版本 API 支持V1 与 V2 的区别和使用场景Stripe Go 多版本 API 支持V1 与 V2 的区别和使用场景 Stripe Go 是一个功能强大的 Go 语言库专为 Stripe API 设计后端物联网消息队列通信EMQX Schema Registry 删除保护拒绝删除被 Schema Validation 或 Message Transformation 引用的内部 SchemaEMQX Schema Registry 删除保护拒绝删除被 Schema Validation 或 Message Transformation 引用的内部后端物联网消息队列通信创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
相关推荐
Phoenix Python Tracing 接入指南：用 arize-phoenix-otel 在 5 分钟内完成 LLM 应用追踪配置
Phoenix Python Tracing 接入指南：用 arize-phoenix-otel 在 5 分钟内完成 LLM 应用追踪配置 【免费下载链接】phoenix AI Observability & Evaluation 项目地址: https://gitcode.com/gh_mirrors/phoenix13/phoenix 本篇指南以 Phoenix 仓库中 phoenix-… · 2026/9/23 19:04:51
SG3525驱动电路设计避坑指南：输出电路与死区时间详解
简介：围绕SG3525电压型PWM控制器的功能特性与典型应用展开，内容覆盖芯片引脚排列、内部构造、电压模式控制原理、软启动与关断电路，以及振荡器充放电时间与频率计算方法，适合开关电源和电机调速方向的工程师、学生作为设计参考。资… · 2026/9/23 19:04:35
二阶有源低通滤波器设计：从理论推导到LM324N焊板避坑指南
简介：这份PDF资料面向电子信息、通信工程等专业的本科生与课程设计学习者，系统整理二阶有源低通滤波器的设计流程与实现方法，帮助读者理解滤波器从理论推导到电路落地的完整思路。内容围绕截止频率10kHz的设计题目展开，涵盖压控电… · 2026/9/23 19:04:35
Java+Vue+Milvus：语义检索与相似文档查重系统实现指南
简介：一套基于Java与Vue的向量数据库语义检索与相似文档查重系统项目详解，面向具备Java和Vue基础的软件工程师、系统架构师，以及从事NLP、知识管理、内容安全的技术人员，解决语义层面改写、同义替换造成的查重漏判问题。项目集成B… · 2026/9/23 20:57:26
166002手写实现：解决版本升级后API全变的痛点
166002手写实现：解决版本升级后API全变的痛点 版本升级后 API 全变了，项目直接崩盘。 别慌，咱们今天不背文档，直接上手。 通过【166002】的手写实现，彻底搞懂底层逻辑。 入口定位：为什么老代码跑不动 很多兄弟在 CSDN… · 2026/9/23 20:57:19
网络电话软件哪个好?实战项目性能优化指南
网络电话软件哪个好?实战项目性能优化指南 看了一堆教程还是不会写项目？别急，今天咱们不聊虚的，直接上手解决【网络电话软件哪个好】背后的性能难题。很多开发者在搭建VoIP系统时，总以为选个开源框架就能跑，结果一上线就卡死、延迟高、掉线频发。这… · 2026/9/23 20:57:19
Semtech推出SurgeSwitch系列新品保护协作机器人免受浪涌与静电放电侵害
专为24V直流工业机器人应用打造、采用极致小巧1.6mm封装的TDS2621LP在0 24A全电流区间内保持近乎恒定的钳位电压专注于为AI数据中心网络和全球智能互联及物联网（IoT）设备提供高性能半导体解决方案的领先企业升特半导体（Semtech，纳… · 2026/9/23 20:57:12
PHP-CS-Fixer 的 return_to_yield_from 规则：将 iterable 函数的 return 数组自动改写为 yield from
PHP-CS-Fixer 的 return_to_yield_from 规则：将 iterable 函数的 return 数组自动改写为 yield from 【免费下载链接】PHP-CS-Fixer A tool to automatically fix PHP Coding Standards issues 项目地址: https://gitcode.com/gh_mirrors/ph/PHP-CS-Fixer 导… · 2026/9/23 20:57:05
5个配置陷阱：新手避坑指南，彻底搞懂冲向炮火源码解析
5个配置陷阱：新手避坑指南，彻底搞懂冲向炮火源码解析 配置环境就卡半天，这大概是每个刚接触《冲向炮火》（Fireworks，指代某高并发网络代理或网关中间件，此处以典型的高性能代理网关源码为原型进行解析，因其架构极具代表性）的新手最真实的写… · 2026/9/23 20:57:05
3招搞定手机怎么下载微信面试难题实战项目解析
3招搞定手机怎么下载微信面试难题实战项目解析 面试被问“手机怎么下载微信”背后的原理，90%的人答不上来。别笑，这看似弱智的问题，实则是考察你对移动应用分发机制、安全校验及网络协议理解的试金石。我带过不少校招新人，他们背了八股文，却连一个A… · 2026/9/23 0:00:03
你有新短消息请注意查收：3个新手避坑指南搞定消息系统选型
你有新短消息请注意查收：3个新手避坑指南搞定消息系统选型 面试被问“高并发下如何保证消息不丢失”，你张口就是“用Redis”，结果面试官追问“如果Redis宕机了怎么办”，你瞬间卡壳。这种场景太常见了，很多新手在背八股文时，只记住了技术名词… · 2026/9/23 0:00:29
Win7无线热点配置工具源码解析：解决API失效的3个实战技巧
Win7无线热点配置工具源码解析：解决API失效的3个实战技巧 Win7无线热点配置工具在Win10/11上跑不动？不是你的问题，是版本升级后 API 全变了。很多老项目里的 netsh wlan… · 2026/9/23 0:00:36