Go+Sarama Kafka场景下如何用Protobuf自描述消息实现通用反序列化

核心思路：用Protobuf Any类型封装多类型消息 #

Protobuf的Any类型天生就是用来解决多类型消息统一传输的问题——它会把任意Protobuf消息打包，并附带消息的类型URL，消费端可以根据这个URL动态识别并解析对应的业务消息，完美解决你当前只能处理单一Pixel类型的局限性。

步骤1：修改Protobuf定义，添加通用消息容器 #

首先更新你的pixel.proto，引入官方的any.proto，并定义一个通用的KafkaMessage容器，所有业务消息都通过这个容器发送：

syntax = "proto3";
package saramaprotobuf;

import "google/protobuf/any.proto";

// 通用Kafka消息容器，用来包裹所有业务消息
message KafkaMessage {
  google.protobuf.Any payload = 1;
}

// 原有的Pixel消息保持不变
message Pixel {
  string session_id = 2;
}

// 可以自由添加其他业务消息类型，比如示例的用户行为事件
message UserEvent {
  string user_id = 1;
  string event_type = 2;
}

编译更新后的proto文件（确保已安装Protobuf编译器和Go语言的Protobuf插件）：

protoc --go_out=. pixel.proto

步骤2：修改生产者，将业务消息包装为Any类型 #

生产者不再直接序列化Pixel，而是把它（或其他业务消息）打包到KafkaMessage的payload字段中，统一发送这个容器消息：

import (
  "github.com/Shopify/sarama"
  "github.com/golang/protobuf/proto"
  "github.com/golang/protobuf/ptypes/any"
  "log"
  "os"
  "os/signal"
  "protobuftest/example" // 替换为你编译后的proto包实际路径
  "syscall"
  "time"
)

// 生产者逻辑修改
go func() {
  ticker := time.NewTicker(time.Second)
  for {
    select {
    case t := <-ticker.C:
      // 示例1：发送Pixel消息
      pixel := &example.Pixel{
        SessionId: t.String(),
      }
      // 将Pixel包装为Any类型
      pixelAny, err := any.New(pixel)
      if err != nil {
        log.Fatalln("Failed to wrap Pixel to Any:", err)
      }
      // 封装到通用KafkaMessage容器
      kafkaMsg := &example.KafkaMessage{
        Payload: pixelAny,
      }
      // 序列化容器消息
      msgBytes, err := proto.Marshal(kafkaMsg)
      if err != nil {
        log.Fatalln("Failed to marshal KafkaMessage:", err)
      }
      // 发送到Kafka
      saramaMsg := &sarama.ProducerMessage{
        Topic: topic,
        Value: sarama.ByteEncoder(msgBytes),
      }
      _, _, err = producer.SendMessage(saramaMsg)
      if err != nil {
        log.Fatalln("Failed to send message:", err)
      }
      log.Printf("Sent Pixel: %s", pixel)

      // 示例2：随时发送其他类型消息，比如UserEvent
      userEvent := &example.UserEvent{
        UserId:    "user_123",
        EventType: "login",
      }
      eventAny, err := any.New(userEvent)
      if err != nil {
        log.Fatalln("Failed to wrap UserEvent to Any:", err)
      }
      kafkaMsg2 := &example.KafkaMessage{
        Payload: eventAny,
      }
      msgBytes2, err := proto.Marshal(kafkaMsg2)
      if err != nil {
        log.Fatalln("Failed to marshal KafkaMessage:", err)
      }
      saramaMsg2 := &sarama.ProducerMessage{
        Topic: topic,
        Value: sarama.ByteEncoder(msgBytes2),
      }
      _, _, err = producer.SendMessage(saramaMsg2)
      if err != nil {
        log.Fatalln("Failed to send message:", err)
      }
      log.Printf("Sent UserEvent: %s", userEvent)
    }
  }
}()

步骤3：修改消费者，动态解析不同类型消息 #

消费端先解析通用的KafkaMessage容器，再根据payload的类型URL判断具体业务消息类型并解析：

// 消费者逻辑修改
for {
  select {
  case msg := <-partitionConsumer.Messages():
    // 先解析通用KafkaMessage容器
    kafkaMsg := &example.KafkaMessage{}
    err := proto.Unmarshal(msg.Value, kafkaMsg)
    if err != nil {
      log.Printf("Failed to unmarshal KafkaMessage: %v", err)
      continue
    }

    // 根据类型URL匹配并解析业务消息
    switch kafkaMsg.Payload.TypeUrl {
    case "type.googleapis.com/saramaprotobuf.Pixel":
      pixel := &example.Pixel{}
      err := any.UnmarshalTo(kafkaMsg.Payload, pixel)
      if err != nil {
        log.Printf("Failed to unmarshal Pixel: %v", err)
        continue
      }
      log.Printf("Received Pixel: %s", pixel)
    case "type.googleapis.com/saramaprotobuf.UserEvent":
      userEvent := &example.UserEvent{}
      err := any.UnmarshalTo(kafkaMsg.Payload, userEvent)
      if err != nil {
        log.Printf("Failed to unmarshal UserEvent: %v", err)
        continue
      }
      log.Printf("Received UserEvent: %s", userEvent)
    default:
      log.Printf("Unknown message type: %s", kafkaMsg.Payload.TypeUrl)
      // 可根据业务需求选择忽略、存储原始数据或报警
    }
  case <-signals:
    log.Print("Received termination signal. Exiting.")
    // 记得关闭资源
    if err := producer.Close(); err != nil {
      log.Printf("Failed to close producer: %v", err)
    }
    if err := partitionConsumer.Close(); err != nil {
      log.Printf("Failed to close partition consumer: %v", err)
    }
    if consumer, ok := partitionConsumer.Consumer().(sarama.Consumer); ok {
      if err := consumer.Close(); err != nil {
        log.Printf("Failed to close consumer: %v", err)
      }
    }
    return
  }
}

关键注意事项 #

• 类型URL格式：默认类型URL为type.googleapis.com/<package>.<message>，如果有自定义前缀需求，可以在any.New()时指定，但一般使用默认即可。
• 更灵活的类型匹配：如果不想用switch硬编码类型URL，可以提前通过proto.RegisterType注册所有业务消息类型，然后通过payload.MessageName()动态获取消息名并创建实例解析。
• 新版Protobuf适配：如果你使用的是google.golang.org/protobuf（推荐新版本），API会略有不同——比如any.New换成proto.MarshalAny，any.UnmarshalTo换成proto.UnmarshalAny，但核心思路完全一致。

这样改造后，你的消费端就能轻松处理任意类型的Protobuf消息，后续新增业务消息只需在proto文件中定义类型，生产者打包为Any发送，消费端添加对应分支即可，完全实现通用化处理。

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