在Go语言中往Kafka中发送数据，通常会使用Sarama库

• Sarama简介
• 基本步骤
• 示例代码

Sarama简介

Sarama是一个用Go语言编写的Apache Kafka客户端库，由Shopify公司最初开发，并在后来被IBM接管维护。Sarama库提供了一套完整的Kafka功能支持，包括生产者（Producer）、消费者（Consumer）以及消费者组（Consumer Group）等组件，允许开发者在Go应用程序中轻松地集成和使用Kafka消息系统。

基本步骤

在Go语言中往Kafka中发送数据，通常会使用Sarama库，这是一个为Apache Kafka设计的Go客户端库。以下是使用Sarama库往Kafka发送数据的基本步骤：

1. 安装Sarama库：首先，你需要安装Sarama库。可以使用Go的包管理工具go get来安装：

   go get github.com/IBM/sarama
   

2. 配置Sarama Producer：创建一个Sarama配置对象，并设置必要的属性，例如Kafka broker地址和消息的RequiredAcks（消息确认的副本数量）：

   config := sarama.NewConfig()
   config.Producer.Return.Successes = true
   config.Producer.Partitioner = sarama.NewRandomPartitioner
   config.Producer.RequiredAcks = sarama.WaitForLocal // 等待本地broker确认
   

3. 创建Producer：使用配置对象创建一个Sarama Producer实例，并指定Kafka的broker地址：

   producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
   if err != nil {
       log.Fatalln(err)
   }
   

4. 发送消息：构建一个Sarama ProducerMessage，并指定Topic和消息内容。然后使用Producer发送这个消息：

   msg := &sarama.ProducerMessage{
       Topic: "your_topic",
       Value: sarama.StringEncoder("your_message"),
   }
   partition, offset, err := producer.SendMessage(msg)
   if err != nil {
       log.Fatalln(err)
   }
   fmt.Printf("Message is stored in topic(%s)/partition(%d)/offset(%d)\n", "your_topic", partition, offset)
   

5. 关闭Producer：发送完消息后，关闭Producer以释放资源：

   if err := producer.Close(); err != nil {
       log.Fatalln(err)
   }
   

以上步骤展示了如何使用Sarama库在Go语言中创建一个Kafka Producer，发送一条消息，并正确关闭资源。在实际应用中，你可能需要根据具体需求调整配置参数，例如设置消息的Key、压缩格式、消息的批次大小等。同时，为了提高性能，你可能会选择使用AsyncProducer而不是SyncProducer，这样可以异步地发送消息，不会阻塞调用线程。

示例代码

package main

import (
	"fmt"
	"log"

	"github.com/IBM/sarama"
)

func main() {
	// Kafka集群的broker地址列表
	brokerList := []string{"localhost:9092"}

	// 创建一个配置对象
	config := sarama.NewConfig()
	// 设置Producer所需的确认模式，这里设置为等待所有同步副本确认
	config.Producer.RequiredAcks = sarama.WaitForAll
	// 设置分区器，这里使用随机分区器
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	// 设置消息成功发送时返回
	config.Producer.Return.Successes = true

	// 使用broker地址和配置创建一个同步Producer
	producer, err := sarama.NewSyncProducer(brokerList, config)
	if err != nil {
		log.Fatalf("Failed to create producer: %v", err)
	}
	defer func() {
		if err := producer.Close(); err != nil {
			log.Fatalf("Failed to close producer: %v", err)
		}
	}()

	// 要发送的消息
	message := &sarama.ProducerMessage{
		Topic: "test",
		Value: sarama.StringEncoder("Your message payload here"),
	}

	// 发送消息
	partition, offset, err := producer.SendMessage(message)
	if err != nil {
		log.Fatalf("Failed to send message: %v", err)
	}

	// 打印消息发送详情
	fmt.Printf("Message is stored in topic(%s)/partition(%d)/offset(%d)\n", message.Topic, partition, offset)
}