我试图通过confluent go客户端向kafka推送一些消息,但问题是消息需要以avro格式推送。在java springboot应用程序中也可以很容易地实现这一点。
我有一种预感,如果通过confluent go客户端,这一切都是可能的。虽然我有一个替代方案来通过融合的rest代理推送这些消息,但这将意味着3-4倍的性能损失,我将拒绝这样做。
我试着用goAvro来转换avro中的消息。虽然我在生产时没有得到任何错误,但数据部分不是以avro格式存储的。

avroCodec, err := goavro.NewCodec(schemaString)
if err != nil {
log.Panic(err.Error())
}
appointmentByte,_ := json.Marshal(appointment)
native, _, _ := avroCodec.NativeFromTextual(appointmentByte)
binaryValue, _ := avroCodec.BinaryFromNative(nil, native)
var recordValue []byte
schemaIDBytes := make([]byte, 4)
binary.BigEndian.PutUint32(schemaIDBytes, uint32(id))
recordValue = append(recordValue, byte(0))
recordValue = append(recordValue, schemaIDBytes...)
recordValue = append(recordValue, binaryValue...)
log.Print(recordValue)
key, _ := uuid.NewUUID()
fmt.Print(key.String())
p.Produce(&kafka.Message{
TopicPartition: kafka.TopicPartition{
Topic: &topic, Partition: kafka.PartitionAny},
Key: []byte(key.String()), Value: recordValue}, nil)发布于 2020-01-30 22:43:59
你可以在Github上搜索你的问题的解决方案。它目前不是项目的一部分,但正在开发中
https://github.com/confluentinc/confluent-kafka-go/issues/69
https://stackoverflow.com/questions/59982917
复制相似问题