Python序列化Protobuf后Golang反序列化报错：无法解析无效WireFormat数据

错误原因 #

你的Python代码使用了Confluent Schema Registry的ProtobufSerializer，它会将Protobuf数据封装成包含Schema ID的自定义格式（即使开启use.deprecated.format: True，依然是Confluent的封装格式，不是原生Protobuf二进制）。而Go代码直接调用proto.Unmarshal解析原生Protobuf二进制，两者格式不匹配，导致解析失败。

解决方案 #

方案1：Go端使用Confluent Schema Registry客户端反序列化（推荐） #

使用Confluent官方的Go Schema Registry客户端，匹配Python端的序列化格式：

1. 安装依赖：

go get github.com/confluentinc/confluent-kafka-go/v2/schemaregistry
go get github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/protobuf

2. 修改Go消费代码：

import (
    "context"
    "log"
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/v2/kafka"
    "github.com/confluentinc/confluent-kafka-go/v2/schemaregistry"
    "github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/protobuf"
    pb "your-proto-package-path" // 替换为你的Protobuf生成包路径
)

func main() {
    // 创建Schema Registry客户端
    srClient, err := schemaregistry.NewClient(schemaregistry.NewConfig("http://localhost:8082"))
    if err != nil {
        log.Fatal(err)
    }

    // 创建Protobuf反序列化器（匹配Python端的deprecated格式配置）
    deserializer, err := protobuf.NewDeserializer(srClient, &protobuf.DeserializerConfig{
        UseDeprecatedFormat: true,
    })
    if err != nil {
        log.Fatal(err)
    }

    // 初始化Kafka消费者
    r, err := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{brokerAddress},
        Topic:   topic,
        GroupID: "test-consumer-group",
        // 其他配置
    })
    if err != nil {
        log.Fatal(err)
    }
    defer r.Close()

    for {
        msg, err := r.ReadMessage(context.Background())
        if err != nil {
            log.Fatal(err)
        }

        user := &pb.UserProperties{}
        // 使用Schema Registry反序列化器解析消息
        err = deserializer.DeserializeInto(msg.Value, user)
        if err != nil {
            log.Fatal(err)
        }

        fmt.Printf("解析后的用户数据: %v\n", user)
    }
}

方案2：Python端改为原生Protobuf序列化（不依赖Schema Registry） #

如果不需要Schema Registry的版本管理功能，可以直接用Protobuf原生序列化：

修改Python代码，替换Confluent的ProtobufSerializer为原生序列化逻辑：

# 替换原有的protobuf_serializer和producer_conf部分
def protobuf_value_serializer(value):
    return value.SerializeToString()

producer_conf = {
    'bootstrap.servers': 'localhost:9092',
    'key.serializer': StringSerializer('utf_8'),
    'value.serializer': protobuf_value_serializer  # 使用原生序列化函数
}
producer = SerializingProducer(producer_conf)

# 后续生产逻辑不变

此时Go端原有的proto.Unmarshal(msg.Value, user)代码即可正常工作，因为数据是原生Protobuf二进制格式。

额外验证点 #

• 确保Python和Go使用的Protobuf定义完全一致（字段名、类型、tag值、嵌套结构均不能有差异）
• 检查Python的protobuf库版本和Go的protoc版本是否兼容（建议使用相同大版本，比如v3.x）

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