Python中如何将Kafka消费的字节数组反序列化为Protobuf对象

• 抛出Unknown magic byte异常的直接原因：Confluent 提供的Protobuf反序列化器有固定格式校验，要求序列化后的消息必须带5字节特殊头部（1字节magic byte标记 + 4字节Schema Registry中对应的schema ID），如果生产者端没有对接Confluent Schema Registry，只是将Protobuf对象序列化为纯字节数组发送，直接使用Confluent反序列化器会因为校验不到合法头部直接报错。
• 你当前的消费者代码没有配置Protobuf对应的反序列化逻辑，拉取到的message.value是原始字节数组，无法直接映射为Protobuf对象，且代码中存在无效导入from email import message，会和Kafka消息对象产生命名冲突。

根据生产者的序列化实现二选一即可：

场景1：生产者未对接Confluent Schema Registry，原生发送Protobuf序列化字节 #

这是适配你当前场景的方案，不需要引入Confluent系列依赖，直接使用Protobuf原生反序列化能力即可，修正后的代码如下：

from kafka import KafkaConsumer
# 移除无效导入：email.message、json、simple模块，避免命名冲突和冗余依赖
import check_pb2

# 替换为你.proto文件中定义的消息类名，例如proto中写的是message SimpleCheck就填SimpleCheck
TARGET_PROTO_CLASS = check_pb2.你的Protobuf消息类名

consumer = KafkaConsumer(
    'latest',
    api_version=(0, 10, 1),
    group_id='my-group',
    enable_auto_commit=False,
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest'
)

for raw_msg in consumer:
    # 实例化Protobuf消息对象
    parsed_msg = TARGET_PROTO_CLASS()
    # 用Kafka消息的原始value字节填充对象
    parsed_msg.ParseFromString(raw_msg.value)
    # 后续直接通过属性访问即可拿到对应字段值，例如parsed_msg.user_id、parsed_msg.content
    print("反序列化得到的Protobuf对象：", parsed_msg)
    # 手动提交消费偏移量
    consumer.commit()

注意事项：

• 消费端使用的check_pb2.py必须和生产者端序列化时使用的.proto文件版本完全一致，否则会出现字段错位、字段值丢失的问题
• 如果生产者发送前对Protobuf序列化后的字节做了额外包装（比如加自定义请求头、Base64编码），需要先拆掉包装层得到原始Protobuf字节，再传入ParseFromString方法

场景2：生产者已对接Confluent Schema Registry #

该场景下需要正确配置Schema Registry连接参数使用Confluent反序列化器，参考代码如下：

from confluent_kafka import DeserializingConsumer
from confluent_kafka.schema_registry.protobuf import ProtobufDeserializer
import check_pb2

TARGET_PROTO_CLASS = check_pb2.你的Protobuf消息类名
# 初始化反序列化器
value_deserializer = ProtobufDeserializer(
    TARGET_PROTO_CLASS,
    {"use.deprecated.format": False}
)

consumer_config = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "my-group",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
    "value.deserializer": value_deserializer,
    # 必须配置你部署的Schema Registry服务地址
    "schema.registry.url": "http://localhost:8081"
}

consumer = DeserializingConsumer(consumer_config)
consumer.subscribe(["latest"])

while True:
    raw_msg = consumer.poll(timeout=1.0)
    if raw_msg is None:
        continue
    if raw_msg.error():
        print(f"消费异常：{raw_msg.error()}")
        continue
    # 此处raw_msg.value直接为解析完成的Protobuf对象
    print("反序列化得到的Protobuf对象：", raw_msg.value)
    consumer.commit(raw_msg)

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