后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载EMQX 的 Schema Registryapps/emqx_schema_registry支持以 Protobuf、Avro、JSON 等格式定义编解码 Schema并在规则引擎中通过schema_encode/schema_decode等 SQL 函数对消息载荷进行转换。本文以仓库变更记录 changes/ee/fix-15839.en.md 为线索完整还原Protobuf Schema 使用map_, _映射字段时编码失败这一问题的现象、根因与修复后的正确用法并结合源码与测试用例emqx_schema_registry_serde_SUITE说明底层实现原理帮助读者在规则引擎中正确使用带 map 字段的 Protobuf 编解码。问题现象带 map 字段的 Protobuf Schema 编码报错在 EMQX 规则引擎中schema_encode函数按注册的 Schema 将传入的 JSON 结构编码为二进制消息。若 Protobuf Schema 中定义了map_, _字段编码会失败。复现所用 Schemaproto3 语法syntax proto3; message test { mapstring, string args 1; }规则 SQL 中使用该 Schema此处 Schema 注册名为xxx消息类型为testSELECT schema_encode(xxx, json_decode(payload), test) as protobuf_test FROM t/#对如下 JSON 载荷进行编码{ args: { env: stag } }会抛出类似如下的错误2025-06-17T06:59:22.72578500:00 [warning] tag: RULE_SQL_EXEC, clientid: c_emqx, msg: SELECT_clause_exception, reason: {error,{gpb_type_error,{bad_unicode_string,[{value,env},{path,test.args.key}]}},[{$schema_parser_xxx,mk_type_error,3,[{file,$schema_parser_xxx.erl},{line,437}]},{$schema_parser_xxx,-v_mapstring,string/3-lc$^0/1-0-,3,[{file,$schema_parser_xxx.erl},{line,429}]},{$schema_parser_xxx,v_mapstring,string,3,[{file,$schema_parser_xxx.erl},{line,429}]},{$schema_parser_xxx,v_msg_test,3,[{file,$schema_parser_xxx.erl},{line,404}]},{$schema_parser_xxx,encode_msg,3,[{file,$schema_parser_xxx.erl},{line,73}]},{emqx_schema_registry_serde,with_serde,2,[{file,emqx_schema_registry_serde.erl},{line,212}]}...注意该 issue 对应仓库中的修复记录为 changes/ee/fix-15839.en.md修复后同一用法可正常编码具体验证方式见下文测试用例验证一节。错误日志剖析gpb_type_error与bad_unicode_string逐段解读上述错误堆栈可以定位到失败点根因元组{gpb_type_error, {bad_unicode_string, [{value, env}, {path, test.args.key}]}}gpb 在编码test.args这个 map 时把 key值为env按 unicode 字符串校验该校验失败。堆栈中的$schema_parser_xxx是 Schema Registry 为注册的 Protobuf Schema 动态生成的编解码模块见下文源码原理v_mapstring,string、v_msg_test、encode_msg分别是 map 字段校验、消息校验和消息编码函数。emqx_schema_registry_serde:with_serde/2是规则 SQL 函数进入编解码的统一入口对应源码 apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl。从根因看失败本质是Protobuf 的 map 字段在 gpb 代码生成时map 的 key 被生成为 unicode 字符串校验逻辑而 EMQX 规则引擎传入的 key 是二进制binary数据二者类型不匹配于是校验函数mk_type_error抛出bad_unicode_string。根因与修复方向maps_key_type编译选项修复的关键在于控制 gpb 生成代码时 map key 的数据类型。在 Schema Registry 生成 Protobuf 编解码模块时统一传入一组编译选项见 apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl 中的base_protobuf_opts/0base_protobuf_opts() - [ binary, strings_as_binaries, descriptor, {maps, true}, {maps_key_type, binary}, {maps_oneof, flat}, {verify, always}, {maps_unset_optional, omitted} ].其中与本问题直接相关的两个选项{maps, true}启用 gpb 对map_, _字段的原生支持将 map 编译为 Erlang map 结构{maps_key_type, binary}指定 map 的 key 统一采用binary类型而不是默认的可能按 unicode charlist/字符串处理的key 类型。源码中 protobuf_cache_key/2 的注释也特别说明Need to take options into account, as changing them might, for example, change the type of map keys between versions编译选项需要计入缓存指纹因为选项变化可能改变 map key 的类型这印证了maps_key_type是决定 map key 表示方式的关键配置。当传入的 map key 为二进制、而生成的代码按 unicode 字符串charlist校验时就会复现 issue 中的bad_unicode_string错误修复后的行为是让 map key 按 binary 处理与规则引擎中json_decode(payload)产出的 binary key 保持一致。修复后的正确用法可运行示例修复后仍使用原来的 Schema、SQL 与载荷即可正常编码通过 Dashboard 或 HTTP API 在 Schema Registry 注册上述test消息类型选择 Protobuf源码即为前文的 proto3 定义在规则引擎中创建规则SQL 保持SELECT schema_encode(xxx, json_decode(payload), test) as protobuf_test FROM t/#向主题t/#发布 JSON 载荷{args: {env: stag}}规则动作即可获得protobuf_test字段其值为按test消息编码后的二进制数据args中的键值对env stag被正确写入 map 字段。编码后的二进制可直接经 MQTT 发布给对端或作为其他桥接动作的载荷。测试用例验证map 类型与 oneof 的往返编解码仓库中的单元测试直接覆盖了本修复场景见 apps/emqx_schema_registry/test/emqx_schema_registry_serde_SUITE.erl 的t_protobuf_map_types/1t_protobuf_map_types(_Config) - Source iolist_to_binary([ [ message test {, mapstring, string args 1;, }, message union {, oneof u {, int32 a 1;, string b 2;, }, } ] ]), Params #{type protobuf, source Source}, SerdeName maps, ok emqx_schema_registry:add_schema(SerdeName, Params), ExtraArgs0 [test], Original0 #{args #{hello world}}, assert_roundtrip(SerdeName, Original0, ExtraArgs0, ExtraArgs0), ...该用例注册一个包含mapstring, string args字段的 Schema并以#{args #{hello world}}binary key/binary value做 encode→decode 往返断言同时覆盖oneof字段的往返编解码。运行该套件emqx_schema_registry_serde_SUITE即可回归验证 map 字段修复。同套件中的t_protobuf_invalid_schema/1emqx_schema_registry_serde_SUITE.erl还验证了非法 Protobuf 源码会以{error, {post_config_update, _, {invalid_protobuf_schema, _}}}的形式被拒注册阶段的源码校验由 emqx_schema_registry.erl 与配置模块 emqx_schema_registry_config.erl 协作完成。源码原理Protobuf 编解码模块的生成与缓存理解该修复还需了解 EMQX Schema Registry 对 Protobuf 的处理机制均在 apps/emqx_schema_registry/src/emqx_schema_registry_serde.erl规则 SQL 函数rsf_schema_encode/1L127-L134与rsf_schema_decode/1分别对应 SQL 中的schema_encode、schema_decode编码结果统一iolist_to_binary/1转为二进制避免下游动作误按 JSON 列表处理。动态代码生成make_protobuf_serde_mod/2L489-L507调用 gpb 的gpb_compile:string/3将用户提交的 Protobuf 源码编译成名为$schema_parser_SchemaName的模块见protobuf_serde_mod_name/1再通过code:load_binary/3装载错误日志堆栈中的$schema_parser_xxx.erl即由此而来。编译选项base_protobuf_opts/0L629-L639统一约束生成的代码风格map 相关行为由{maps, true}与{maps_key_type, binary}决定。编译缓存为避免多节点重复编译lazy_generate_protobuf_code/3L546-L557在 mria 事务内加锁执行编译结果按{SchemaName, OTP版本, MD5指纹}缓存于 mnesia 表?PROTOBUF_CACHE_TAB指纹protobuf_cache_key/2包含编译选项哈希与全部源码内容因此修改源码或升级 OTP 都会触发重新编译。销毁清理删除 Schema 时通过destroy_protobuf_code/1L713-L722卸载模块并删除缓存条目测试用例t_destroy_protobuf/1与t_update_protobuf_cache/1L251-L307分别验证了缓存命中/失效与销毁行为。在规则引擎中使用 Protobuf 编解码的建议结合本次修复实践中请注意以下几点map 字段的键值类型规则 SQL 中json_decode(payload)产出的 JSON 对象键为二进制字符串与maps_key_type binary的生成代码对齐应避免在载荷中使用非字符串类型的 map key如数字 key否则 gpb 仍可能报类型错误。区分 encode 与 decodeschema_encode(SchemaName, Term, MessageType)输入 JSON 结构、输出二进制schema_decode(SchemaName, Binary, MessageType)反之。若把已解码的 map 再传给schema_decode会触发 eval_decode/2 中的显式schema_decode_error提示Attempted to schema decode an already decoded message。Schema 变更需重新注册修改 Protobuf 源码后缓存指纹源码 MD5变化会触发重新编译无需重启节点但应通过 Schema Registry 的更新接口重新提交避免旧模块残留相关缓存清理逻辑见t_update_protobuf_cache用例。验证手段可参考 emqx_schema_registry_serde_SUITE.erl 中的assert_roundtrip/3模式先 encode 再 decode 对比原始结构快速确认 Schema 定义是否符合预期。小结本文从变更记录 changes/ee/fix-15839.en.md 出发完整还原了Protobuf Schema 的map_, _字段编码失败问题其根因是 gpb 生成代码时 map key 的类型处理与规则引擎传入的二进制 key 不一致导致gpb_type_error: bad_unicode_string修复通过编译选项{maps_key_type, binary}统一 map key 为二进制类型。该修复已有回归测试覆盖emqx_schema_registry_serde_SUITE.erl相关实现集中在 emqx_schema_registry_serde.erl。在规则引擎中编写含 map 字段的 Protobuf Schema 时保持 JSON 键为字符串并遵循 encode/decode 的输入输出约定即可稳定完成编解码。赞分享后端物联网消息队列通信【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址https://gitcode.com/gh_mirrors/em/emqx点击查看免费下载相关推荐EMQX Schema Registry 实战指南在规则引擎中统一管理 Avro / Protobuf / JSON Schema 编解码EMQX Schema Registry 实战指南在规则引擎中统一管理 Avro / Protobuf / JSON Schema 编解码 Schema Re后端物联网消息队列通信Buzz 如何识别转录中的说话人并修改说话人标签Buzz 如何识别转录中的说话人并修改说话人标签 Buzz 可以对已生成转录的音频或视频文件做说话人识别把每位说话人的句子标上标签允许你把自动生成的标签后端物联网消息队列通信EMQX 规则引擎租户命名空间下的全局规则匹配修复limit_selects_in_namespace 机制与源码剖析EMQX 规则引擎租户命名空间下的全局规则匹配修复 limit_selects_in_namespace 机制与源码剖析 本篇文章围绕 EMQX 变更记录 f后端物联网消息队列通信创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考