将cloudevents格式的数据发送到Kafka主题

从现在开始，我们要努力学习啦！今天我给大家带来《将cloudevents格式的数据发送到Kafka主题》，感兴趣的朋友请继续看下去吧！下文中的内容我们主要会涉及到等等知识点，如果在阅读本文过程中有遇到不清楚的地方，欢迎留言呀！我们一起讨论，一起学习！

现在我有这段代码并且工作正常。 （它向kafka主题发送一些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)

但有人希望拥有 cloudevents 格式的数据。 -> https://github.com/cloudevents/sdk-go 那么我该怎么办，因为这个event结构不能直接转换为字符串。

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)

我应该怎么做才能将此事件发送到 kafka？尝试将 event.dataencoded 转换为字符串或类似的东西？ 顺便提一句。 编程语言是golang。

正确答案

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

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)

...

今天关于《将cloudevents格式的数据发送到Kafka主题》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！