Golang怎么连接Kafka

Kafka是一个分布式流处理平台，广泛用于构建实时数据管道和流应用。Golang作为一种高效、简洁的编程语言，与Kafka的结合可以构建高性能的实时数据处理系统。本文将介绍如何使用Golang连接Kafka，并进行消息的生产和消费。

1. 安装Kafka客户端库

在Golang中，常用的Kafka客户端库是confluent-kafka-go，它是基于librdkafka的Golang绑定。首先，我们需要安装这个库。

go get -u github.com/confluentinc/confluent-kafka-go/kafka

2. 配置Kafka生产者和消费者

2.1 生产者配置

生产者负责将消息发送到Kafka的指定主题。以下是一个简单的生产者配置示例：

package main

import (
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
	producer, err := kafka.NewProducer(&kafka.ConfigMap{
		"bootstrap.servers": "localhost:9092",
	})
	if err != nil {
		panic(err)
	}
	defer producer.Close()

	topic := "test-topic"
	message := "Hello, Kafka!"
	producer.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          []byte(message),
	}, nil)

	// 等待消息发送完成
	producer.Flush(15 * 1000)
	fmt.Println("Message sent successfully")
}

2.2 消费者配置

消费者负责从Kafka的指定主题中读取消息。以下是一个简单的消费者配置示例：

package main

import (
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
	consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": "localhost:9092",
		"group.id":          "my-group",
		"auto.offset.reset": "earliest",
	})
	if err != nil {
		panic(err)
	}
	defer consumer.Close()

	topic := "test-topic"
	consumer.SubscribeTopics([]string{topic}, nil)

	for {
		msg, err := consumer.ReadMessage(-1)
		if err == nil {
			fmt.Printf("Received message: %s\n", string(msg.Value))
		} else {
			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
		}
	}
}

3. 运行示例

3.1 启动Kafka

在运行Golang代码之前，确保Kafka服务已经启动。可以使用以下命令启动Kafka：

# 启动Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# 启动Kafka
bin/kafka-server-start.sh config/server.properties

3.2 运行生产者和消费者

分别运行生产者和消费者代码：

# 运行生产者
go run producer.go

# 运行消费者
go run consumer.go

如果一切正常，生产者将发送消息到Kafka的test-topic主题，消费者将从该主题中读取并打印消息。

4. 总结

本文介绍了如何使用Golang连接Kafka，并通过简单的示例代码展示了如何配置和使用Kafka的生产者和消费者。通过这些基础操作，您可以进一步构建更复杂的实时数据处理系统。希望本文对您有所帮助！