Go 如何通过 Kafka 客户端库 生产与消费消息

在Go中，你可以使用sarama库来作为Apache Kafka的客户端进行生产和消费消息。以下是一个简单的生产者和消费者的例子。

首先，通过以下命令安装sarama库：

go get github.com/Shopify/sarama

生产者代码示例：

package main
 
import (
    "fmt"
    "github.com/Shopify/sarama"
)
 
func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    if err != nil {
        fmt.Println("Failed to start producer:", err)
        return
    }
 
    msg := &sarama.ProducerMessage{
        Topic: "myTopic",
        Value: sarama.StringEncoder("Hello Kafka!"),
    }
 
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        fmt.Println("Failed to send message:", err)
        return
    }
 
    fmt.Printf("Message sent to partition %d at offset %d\n", partition, offset)
}

消费者代码示例：

package main
 
import (
    "fmt"
    "github.com/Shopify/sarama"
)
 
func main() {
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
    if err != nil {
        panic(err)
    }
 
    partitionConsumer, err := consumer.ConsumePartition("myTopic", 0, sarama.OffsetNewest)
    if err != nil {
        panic(err)
    }
    defer partitionConsumer.Close()
 
    for msg := range partitionConsumer.Messages() {
        fmt.Printf("Message received: %s\n", msg.Value)
    }
}

在这两个例子中，请确保将"localhost:9092"替换为你的Kafka集群的实际地址，并且将"myTopic"替换为你要生产或消费消息的实际主题。