利用protobuf的消息自描述功能

最近发现不少小伙伴都对Golang很感兴趣，所以今天继续给大家介绍Golang相关的知识，本文《利用protobuf的消息自描述功能》主要内容涉及到等等知识点，希望能帮到你！当然如果阅读本文时存在不同想法，可以在评论中表达，但是请勿使用过激的措辞~

我在使用 protocol buffers 时正在研究的用例之一是反序列化我在消费者端收到的 protocol buffers 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 文件并在主函数中引用它，以便发送和接收消息。这里的问题是，解决方案不是通用的。我会在consumer端收到不同.proto类型的消息。

如何使其通用？我知道有一种叫做自描述消息（动态消息）的东西作为 protobuf 的一部分。我引用了此链接 https://developers.google.com/protocol-buffers/docs/techniques?csw=1#self-description 。但它没有任何解释如何将其嵌入为 pixel.proto 的一部分（我使用过的示例），因此在消费者端我直接将其反序列化为所需的类型。

解决方案

您将定义一个通用容器消息类型，其中包含 DescriptorSet 和 Any 字段。

发送时，您可以构建该通用消息类型的实例，使用 Pixel 消息的实例设置 Any 类型的字段，并使用 Pixel 类型的 DescriptorSet 设置 DescriptorSet 字段。

这将允许此类消息的接收者使用您附加的描述符集来解析任何内容。实际上，这是将一段原始定义与消息一起发送。因此接收者不需要预先共享的原型定义或生成的代码。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《利用protobuf的消息自描述功能》文章吧，也可关注golang学习网公众号了解相关技术文章。