以云事件格式将数据发送到 Kafka 主题

Question

现在我有这个代码，它工作正常。（它向卡夫卡主题发送一些 json 格式的数据）j, err := json.Marshal(data)if err != nil {    log.Fatal(err)}msg := &sarama.ProducerMessage{    Topic: tName,    Value: sarama.StringEncoder(j),}_, _, err = producer.SendMessage(msg)但有人希望以云事件格式拥有这些数据。->https://github.com/cloudevents/sdk-go 所以我该怎么办，因为这个事件结构不能直接转换为字符串。type Event struct {    Context     EventContext    DataEncoded []byte    // DataBase64 indicates if the event, when serialized, represents    // the data field using the base64 encoding.    // In v0.3, this field is superseded by DataContentEncoding    DataBase64  bool    FieldErrors map[string]error}所以这段代码甚至不会编译。j, err := json.Marshal(data)if err != nil {    log.Fatal(err)}//...event := cloudevents.NewEvent()event.SetSource("example/uri") event.SetType("example.type")event.SetData(cloudevents.ApplicationJSON, j)producerMsg := &sarama.ProducerMessage{    Topic: s.outputTopic,    Value: sarama.StringEncoder(event),}_, _, err = s.producer.SendMessage(producerMsg)我应该怎么做才能将此活动发送给卡夫卡？尝试投射事件。数据编码为字符串或类似的东西？顺便说一句。编程语言是戈兰语。

Answer 1

您是否看到序列化事件的文档部分？

https://github.com/cloudevents/sdk-go#serializedeserialize-a-cloudevent

event := cloudevents.NewEvent()

event.SetSource("example/uri") 

event.SetType("example.type") 

// data here is a map[string] interface{}, or some other Struct type representing the "example.type" schema type above 

event.SetData(cloudevents.ApplicationJSON, data)

bytes, err := json.Marshal(event)

if err != nil {

  log.Fatal(err)

}

producerMsg := &sarama.ProducerMessage{

    Topic: s.outputTopic,

    Value: bytes,  // you've already encoded the event 

}

否则，请务必查看提供的使用 CloudEvent 客户端的示例代码 https://github.com/cloudevents/sdk-go/blob/main/samples/kafka/sender/main.go

sender, err := kafka_sarama.NewSender([]string{"127.0.0.1:9092"}, saramaConfig, "test-topic")

if err != nil {

    log.Fatalf("failed to create protocol: %s", err.Error())

}

defer sender.Close(context.Background())

c, err := cloudevents.NewClient(sender, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())

if err != nil {

    log.Fatalf("failed to create client, %v", err)

}

event := cloudevents.NewEvent() 

event.Set... 

c.Send(..., event)

...