慎重声明，只代表本人观点，不一定代表实际。看了差不多半天Spring Cloud Stream中的kafka源代码，差不多断断续续折腾了一个月，终于在golang中使用kafka发送给Spring Cloud Stream并且成功处理
func testKafka() {
sarama.Logger = log.New(os.Stdout, "", log.LstdFlags)
config := sarama.NewConfig()
config.Producer.Return.Successes = true
config.Producer.Return.Errors = true
config.Version = sarama.MaxVersion
client, err := sarama.NewClient(strings.Split("localhost:9092", ","), config)
if err != nil {
log.Fatalf("unable to create kafka client: %q", err)
}
producer, err := sarama.NewSyncProducerFromClient(client)
if err != nil {
log.Fatalf("unable to create producer: %q", err)
}
defer producer.Close()
serviceErrorLog := &ServiceErrorLog{ApplicationName: "test-service", ServerIp: "127.0.0.1", Path: "/", QueryParams: "/QueryParams", Message: "Message", Trace: "Trace", LogTime: time.Now().Format(DATE_TIME_PATTERN)}
if err == nil {
key, err1 := uuid.NewRandom()
if err1 != nil {
log.Fatalf("unable to create uuid: %q", err1)
}
//headers := &MessageHeader{Id: key.String(), ContentType: "application/json", PartitionKey: "service-error-logs", Timestamp: time.Now().UnixNano()}
//genericMessage := &ServiceErrorLogGenericMessage{Headers: *headers, Payload: *serviceErrorLog}
text, _ := json.Marshal(serviceErrorLog)
id := utility.UUIDJavaBytes(key)
contentType := []byte("application/json")
partitionKey := []byte("service-error-logs")
springJsonHeaderTypes := []byte{123, 34, 112, 97, 114, 116, 105, 116, 105, 111, 110, 75, 101, 121, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 115, 99, 115, 116, 95, 112, 97, 114, 116, 105, 116, 105, 111, 110, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 73, 110, 116, 101, 103, 101, 114, 34, 44, 34, 99, 111, 110, 116, 101, 110, 116, 84, 121, 112, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 125}
message := &sarama.ProducerMessage{
Topic: "log-service-topic",
Headers: []sarama.RecordHeader{
{Key: []byte("id"), Value: id},
{Key: []byte("contentType"), Value: contentType},
{Key: []byte("partitionKey"), Value: partitionKey},
{Key: []byte("spring_json_header_types"), Value: springJsonHeaderTypes},
},
Value: sarama.StringEncoder(text)}
fmt.Println(message)
partition, offset, err := producer.SendMessage(message)
fmt.Println(partition, offset, err)
}
}