如何用Python处理Protobuf解析的ZMQ嵌套消息并提取可变数量的指定字段

客户端侧可落地实现方案 #

1. 解决protobuf对象无法JSON序列化的问题 #

直接对ParseFromString()返回的protobuf Message对象调用json.dumps一定会报错，这类对象是封装的原生C++结构，不属于Python基础可序列化类型。直接用protobuf官方内置的转换工具，把对象转成完全原生的Python字典即可从根源解决报错：

from google.protobuf.json_format import MessageToDict
# 假设parse得到的根消息对象是root_msg
raw_python_dict = MessageToDict(
    root_msg,
    preserving_proto_field_name=True, # 保留proto定义的原始字段名，避免自动做驼峰格式转换
)

转换完成后得到的结构全是Python原生的dict、list、数值、字符串类型，后续可以直接做任意取值、序列化操作，不会再触发类型不兼容的问题。

2. 适配拼写差异、动态提取数量不固定的sys类字段 #

sys_activity、sys_activty（proto定义本身的拼写错误，并非示例笔误）都是重复嵌套字段，转成原生字典后是长度不固定的列表，每个列表元素是包含key、value的字典，不需要硬编码固定条目数量，直接遍历全量收集key值即可：

# 定位到some_activity对应的字典层级
some_activity_node = raw_python_dict["summaries"][0]["value"]["some_activity"]
# 初始化收集列表
sys_key_list = []
# 收集拼写正确的sys_activity下的所有key
for item in some_activity_node.get("sys_activity", []):
    sys_key_list.append(item["key"])
# 单独兼容拼写错误的sys_activty下的所有key，避免漏掉条目
for item in some_activity_node.get("sys_activty", []):
    sys_key_list.append(item["key"])
# 直接提取固定位置的事件计数字段
some_events_a = some_activity_node["some_events_A"]
some_events_b = some_activity_node["some_events_B"]

如果直接操作原始protobuf对象的属性，同样可以实现逻辑：直接遍历some_activity.sys_activity、some_activity.sys_activty两个重复字段容器里的每个元素取key属性即可，两种方式的适配逻辑完全一致。

3. 按目标格式组装输出 #

把提取到的字段按照要求拼接成最终格式即可：

final_output = (
    f"some_activity: {sys_key_list},\n"
    f"some_events_A: {some_events_a}\n"
    f"some_events_B: {some_events_b}"
)
print(final_output)

实现注意事项 #

• 所有非空字段取值时，尽量用.get()方法加默认值兜底，避免遇到空的嵌套对象（比如示例中的more_activity、error_activity）时触发KeyError
• 遍历重复字段的逻辑完全不需要适配条目数量变化，就算后续消息里的sys类字段从3个变成几十上百个，也不需要修改代码，自动完成全量提取
• 不需要对服务端的proto定义做任何修改，所有操作都在客户端解析完成后的本地数据上完成

内容的提问来源于stack exchange，提问作者Vik