本文将详细介绍如何在 Sealos DevBox 中使用 Go 语言连接和操作 Kafka。

准备工作

安装依赖

在 Cursor 终端中执行以下命令安装所需的依赖包：

go get github.com/joho/godotenv
go get github.com/confluentinc/confluent-kafka-go/v2/kafka

上述命令将安装：

github.com/confluentinc/confluent-kafka-go/v2/kafka：用于 Go 语言的 Confluent Kafka 客户端
github.com/joho/godotenv：用于加载环境变量配置文件

系统依赖

由于 confluent-kafka-go 包依赖于 librdkafka，您需要在 DevBox 开发环境中安装相关系统依赖。请在 Cursor 终端中执行以下命令：

sudo apt-get update
sudo apt-get install -y gcc libc6-dev librdkafka-dev

配置与实现

配置环境变量

首先，在项目根目录创建 .env 文件，添加 Kafka 连接所需的环境变量：

.env

KAFKA_BROKER=your_kafka_host:9092
KAFKA_GROUP_ID=group-id
KAFKA_TOPIC=topic-name

请将上述配置中的占位符替换为您从 Sealos 数据库应用中获取的实际 Kafka 连接信息。

编写主程序

创建 main.go 文件，实现 Kafka 消息的生产和消费功能：

main.go

package main
 
import (
	"fmt"
	"log"
	"os"
 
	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
	"github.com/joho/godotenv"
)
 
var (
	broker  string
	groupId string
	topic   string
)
 
func loadEnv() error {
	// Load environment variables from .env file
	err := godotenv.Load()
	if err != nil {
		log.Fatal("Error loading .env file")
	}
 
	broker = os.Getenv("KAFKA_BROKER")
	groupId = os.Getenv("KAFKA_GROUP_ID")
	topic = os.Getenv("KAFKA_TOPIC")
 
	return nil
}
 
func startProducer() {
	p, err := kafka.NewProducer(&kafka.ConfigMap{
		"bootstrap.servers":        broker,
		"allow.auto.create.topics": true,
	})
	if err != nil {
		panic(err)
	}
 
	go func() {
		for e := range p.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
				} else {
					fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
				}
			}
		}
	}()
 
	for _, word := range []string{"message 1", "message 2", "message 3"} {
		p.Produce(&kafka.Message{
			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
			Value:          []byte(word),
		}, nil)
	}
}
 
func startConsumer() {
	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": broker,
		"group.id":          groupId,
		"auto.offset.reset": "earliest",
	})
 
	if err != nil {
		panic(err)
	}
	c.Subscribe(topic, nil)
 
	for {
		msg, err := c.ReadMessage(-1)
		if err == nil {
			fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
		} else {
			fmt.Printf("Consumer error: %v (%v)\n", err, msg)
			break
		}
	}
 
	c.Close()
}
 
func main() {
	if err := loadEnv(); err != nil {
		fmt.Println(err)
		return
	}
 
	startProducer()
	startConsumer()
}

代码主要包含以下几个部分：

导入和变量定义：导入必要的包，并定义 Kafka 连接所需的全局变量。
生产者实现 (startProducer 函数)：
- 创建 Kafka 生产者实例
- 使用 goroutine 处理消息投递状态
- 向指定主题发送示例消息
消费者实现 (startConsumer 函数)：
- 创建 Kafka 消费者实例
- 订阅指定主题
- 持续读取并处理消息
主函数：加载环境变量并启动生产者和消费者，演示完整的消息处理流程。

运行程序

在运行环境终端中执行以下命令启动程序：

go run main.go

程序将演示 Kafka 消息的生产和消费过程。

最佳实践

在生产环境中，建议将生产者和消费者分别部署为独立的服务。
使用环境变量管理配置信息，避免硬编码。
实现完善的错误处理和日志记录机制。
确保程序能够优雅关闭，正确释放 Kafka 连接资源。

常见问题

如果遇到连接问题，请检查以下几点：

确认 Kafka 集群地址配置是否正确。
验证 Kafka 集群是否正常运行且可访问。
检查运行环境的网络配置是否有限制。
确保所有依赖包和系统组件已正确安装。
如果遇到 cgo 相关错误，请安装必要的构建工具 (执行 sudo apt-get install build-essential)。

更多关于 Go 语言操作 Kafka 的详细信息，请参考 confluent-kafka-go 官方文档。

Go

本页导航