在 Go (Golang) 中使用 Kafka：segmentio/kafka-go 库

在 Go (Golang) 中使用 segmentio/kafka-go 库操作 Kafka

Apache Kafka 是一个分布式流数据处理平台，已成为构建高负载消息交换系统的事实标准。在使用 Go (Golang) 开发时，最流行且可靠的 Kafka 交互库之一是 segmentio/kafka-go。本文将详细介绍如何安装、配置并使用该库来创建生产者和消费者。

为什么选择 segmentio/kafka-go？

市场上有多个用于 Kafka 的 Go 库（例如 Sarama 和 Confluent Go Client）。segmentio/kafka-go 具有以下优势：

• 纯 Go 实现 — 不需要 C 依赖（与 Confluent 不同），简化了构建和部署。
• 支持 Kafka Protocol v2 — 高效的压缩、批处理和事务处理。
• 简单的 API — 为生产者、消费者和主题管理提供直观的接口。
• 活跃的社区 — 定期更新和良好的文档。

安装库

首先，通过 go get 将库添加到您的项目中：

go get github.com/segmentio/kafka-go

确保您已启动 Kafka 实例（本地或通过 Docker）。最小 Docker Compose 配置示例：

version: '3'services:  zookeeper:    image: confluentinc/cp-zookeeper:latest    environment:      ZOOKEEPER_CLIENT_PORT: 2181  kafka:    image: confluentinc/cp-kafka:latest    depends_on:      - zookeeper    ports:      - "9092:9092"    environment:      KAFKA_BROKER_ID: 1      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

创建生产者 (Producer)

生产者将消息发送到 Kafka 主题。基本示例：

package main

import (    "context"    "fmt"    "log"    "time"

    "github.com/segmentio/kafka-go")

func main() {    // 配置生产者    writer := &kafka.Writer{        Addr:     kafka.TCP("localhost:9092"),        Topic:    "test-topic",        Balancer: &kafka.LeastBytes{}, // 按消息大小进行负载均衡        BatchSize: 10,                 // 每批发送 10 条消息        BatchTimeout: 100 * time.Millisecond,    }    defer writer.Close()

    // 发送消息    err := writer.WriteMessages(context.Background(),        kafka.Message{            Key:   []byte("key-1"),            Value: []byte("Hello, Kafka from Go!"),        },    )    if err != nil {        log.Fatal("发送消息失败:", err)    }

    fmt.Println("消息发送成功")}

kafka.Writer 的关键参数：

• Addr — Kafka 代理地址。
• Topic — 主题名称。
• Balancer — 消息分区分配策略（LeastBytes、RoundRobin、Hash）。
• BatchSize 和 BatchTimeout — 性能优化参数。

创建消费者 (Consumer)

消费者从主题中读取消息。手动管理 offset 的简单示例：

package main

import (    "context"    "fmt"    "log"

    "github.com/segmentio/kafka-go")

func main() {    // 配置读取器（消费者）    reader := kafka.NewReader(kafka.ReaderConfig{        Brokers:   []string{"localhost:9092"},        Topic:     "test-topic",        GroupID:   "my-consumer-group",        MinBytes:  10, // 最小批处理大小        MaxBytes:  10e6, // 10 MB        StartOffset: kafka.FirstOffset, // 从