go - 如何为 protbuf 使用自描述消息
问题描述
我在使用协议缓冲区时正在处理的用例之一是反序列化我在消费者端收到的协议缓冲区 Kafka 消息(使用 sarama 库和 Go)。
我目前的做法是我定义了示例 pixel.proto 文件,如下所示。
syntax = "proto3";
package saramaprotobuf;
message Pixel {
// Session identifier stuff
string session_id = 2;
}
我正在通过 sarama.Producer 发送消息(通过编组)接收它 sarama.Consumer (通过引用已编译的 pixel.proto.pb 来解组消息)。代码如下。
import (
"github.com/Shopify/sarama"
"github.com/golang/protobuf/proto"
"log"
"os"
"os/signal"
"protobuftest/example"
"syscall"
"time"
)
func main() {
topic := "test_topic"
brokerList := []string{"localhost:9092"}
producer, err := newSyncProducer(brokerList)
if err != nil {
log.Fatalln("Failed to start Sarama producer:", err)
}
go func() {
ticker := time.NewTicker(time.Second)
for {
select {
case t := <-ticker.C:
elliot := &example.Pixel{
SessionId: t.String(),
}
pixelToSend := elliot
pixelToSendBytes, err := proto.Marshal(pixelToSend)
if err != nil {
log.Fatalln("Failed to marshal example:", err)
}
msg := &sarama.ProducerMessage{
Topic: topic,
Value: sarama.ByteEncoder(pixelToSendBytes),
}
producer.SendMessage(msg)
log.Printf("Pixel sent: %s", pixelToSend)
}
}
}()
signals := make(chan os.Signal, 1)
signal.Notify(signals, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM)
partitionConsumer, err := newPartitionConsumer(brokerList, topic)
if err != nil {
log.Fatalln("Failed to create Sarama partition consumer:", err)
}
log.Println("Waiting for messages...")
for {
select {
case msg := <-partitionConsumer.Messages():
receivedPixel := &example.Pixel{}
err := proto.Unmarshal(msg.Value, receivedPixel)
if err != nil {
log.Fatalln("Failed to unmarshal example:", err)
}
log.Printf("Pixel received: %s", receivedPixel)
case <-signals:
log.Print("Received termination signal. Exiting.")
return
}
}
}
func newSyncProducer(brokerList []string) (sarama.SyncProducer, error) {
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Retry.Max = 5
config.Producer.Return.Successes = true
// TODO configure producer
producer, err := sarama.NewSyncProducer(brokerList, config)
if err != nil {
return nil, err
}
return producer, nil
}
func newPartitionConsumer(brokerList []string, topic string) (sarama.PartitionConsumer, error) {
conf := sarama.NewConfig()
// TODO configure consumer
consumer, err := sarama.NewConsumer(brokerList, conf)
if err != nil {
return nil, err
}
partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetOldest)
if err != nil {
return nil, err
}
return partitionConsumer, err
}
如您所见,在代码中,我已导入 .proto 文件并在主函数中引用它,以便发送和接收消息。这里的问题是,解决方案不是通用的。我会在消费者端收到不同的 .proto 类型的消息。
我怎样才能使它通用?我知道作为 protobuf 的一部分,有一种叫做自描述消息(动态消息)的东西。我提到了这个链接https://developers.google.com/protocol-buffers/docs/techniques?csw=1#self-description。但它没有任何解释如何将其嵌入为 pixel.proto 的一部分(我使用过的示例),以便在消费者端我直接将其反序列化为所需的类型。
解决方案
您将定义一个包含 DescriptorSet 和 Any 字段的通用容器消息类型。
发送时,您构建该通用消息类型的实例,使用 Pixel 消息的实例设置 Any 类型的字段,并使用 Pixel 类型的 DescriptorSet 设置 DescriptorSet 字段。
这将允许此类消息的接收者使用您附加的 DescriptorSet 解析 Any 内容。实际上,这是与消息一起发送一段 proto 定义。所以接收者不需要预先共享的原型定义或生成的代码。
话虽如此,我不确定这是否是您真正想要的,因为如果您打算与客户共享原型定义或生成的代码,那么我建议在容器类型中简单地使用 oneof 字段会更容易使用。
推荐阅读
- mongodb - MONGODB 聚合查询以按元素给出数组字段的 $min 值
- node.js - 在 Dockerfile 中基于 nvm 的 LTS 安装后 npm 安装失败
- python - 重塑和重命名 Pandas DataFrame 列轴
- javascript - Webpack 找不到任何导入的模块
- node.js - 为什么Mongo数据库中的数据和我从flutter App输入的数据不一样?
- mqtt - mqtt 知道队列何时被处理?
- c++ - 在 Visual Studio 2019 中使用枚举作为模板参数时出现错误 C2440 和 C2973
- elasticsearch - 我可以从 Elasticsearch 中的嵌套键值中过滤文档吗?
- python - 在python中发送http请求的最快方法
- python - Python:替换整个json中嵌套的dict键的值,无论是在列表中还是直接在dict中