企业官网建设流程全解析

• 后端
• 物联网
• 消息队列
• 通信

EMQX 的 Schema Registry（apps/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<_, _>字段，编码会失败。

复现所用 Schema（proto3 语法）：

syntax = "proto3"; message test { map<string, string> args = 1; }

规则 SQL 中使用该 Schema（此处 Schema 注册名为xxx，消息类型为test）：

SELECT schema_encode('xxx', json_decode(payload), 'test') as protobuf_test FROM "t/#"

对如下 JSON 载荷进行编码：

{ "args": { "env": "stag" } }

会抛出类似如下的错误：

2025-06-17T06:59:22.725785+00: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_map<string,string>/3-lc$^0/1-0-',3,[{file,"$schema_parser_xxx.erl"},{line,429}]},{'$schema_parser_xxx','v_map<string,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_map<string,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/0：

base_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 与载荷即可正常编码：

1. 通过 Dashboard 或 HTTP API 在 Schema Registry 注册上述test消息（类型选择 Protobuf，源码即为前文的 proto3 定义）；
2. 在规则引擎中创建规则，SQL 保持：

SELECT schema_encode('xxx', json_decode(payload), 'test') as protobuf_test FROM "t/#"

3. 向主题t/#发布 JSON 载荷{"args": {"env": "stag"}}；
4. 规则动作即可获得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/1：

t_protobuf_map_types(_Config) -> Source = iolist_to_binary([ [ "message test {", " map<string, 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), ...

该用例注册一个包含map<string, string> args字段的 Schema，并以#{<<"args">> => #{<<"hello">> => <<"world">>}}（binary key/binary value）做 encode→decode 往返断言，同时覆盖oneof字段的往返编解码。运行该套件（emqx_schema_registry_serde_SUITE）即可回归验证 map 字段修复。

同套件中的t_protobuf_invalid_schema/1（emqx_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/1（L127-L134）与rsf_schema_decode/1分别对应 SQL 中的schema_encode、schema_decode；编码结果统一iolist_to_binary/1转为二进制，避免下游动作误按 JSON 列表处理。
• 动态代码生成：make_protobuf_serde_mod/2（L489-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/0（L629-L639）统一约束生成的代码风格，map 相关行为由{maps, true}与{maps_key_type, binary}决定。
• 编译缓存：为避免多节点重复编译，lazy_generate_protobuf_code/3（L546-L557）在 mria 事务内加锁执行，编译结果按{SchemaName, OTP版本, MD5指纹}缓存于 mnesia 表?PROTOBUF_CACHE_TAB；指纹（protobuf_cache_key/2）包含编译选项哈希与全部源码内容，因此修改源码或升级 OTP 都会触发重新编译。
• 销毁清理：删除 Schema 时通过destroy_protobuf_code/1（L713-L722）卸载模块并删除缓存条目，测试用例t_destroy_protobuf/1与t_update_protobuf_cache/1（L251-L307）分别验证了缓存命中/失效与销毁行为。

在规则引擎中使用 Protobuf 编解码的建议

结合本次修复，实践中请注意以下几点：

1. map 字段的键值类型：规则 SQL 中json_decode(payload)产出的 JSON 对象键为二进制字符串，与maps_key_type => binary的生成代码对齐；应避免在载荷中使用非字符串类型的 map key（如数字 key），否则 gpb 仍可能报类型错误。
2. 区分 encode 与 decode：schema_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"）。
3. Schema 变更需重新注册：修改 Protobuf 源码后，缓存指纹（源码 MD5）变化会触发重新编译，无需重启节点；但应通过 Schema Registry 的更新接口重新提交，避免旧模块残留（相关缓存清理逻辑见t_update_protobuf_cache用例）。
4. 验证手段：可参考 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 的输入输出约定，即可稳定完成编解码。

• 后端
• 物联网
• 消息队列
• 通信