Go 客户端 kafka-go

中间件

Kafka

Go 客户端 kafka-go

使用 github.com/segmentio/kafka-go 操作 Kafka：低级别 Conn API、Topic 创建、高级别 Reader / Writer API。

如果你对 Kafka 的核心概念还不熟悉，建议先阅读 01-核心概念（基本术语部分）。

kafka-go 简介

本文使用 segmentio/kafka-go 作为 Go 客户端。

go get github.com/segmentio/kafka-go

版本要求：Go >= 1.15 兼容性：kafka-go 已在 Kafka 0.10.1.0 到 2.7.1.0 版本上经过测试。虽然更高版本的 Kafka 通常也能兼容，但某些较新的 Kafka API 功能可能尚未在客户端中实现。

两个抽象层级

kafka-go 提供了两个层级的 API：

低级别（Low-Level）：基于 Conn 类型，封装原始网络连接，提供更底层的控制。类比手动挡——灵活度更高，但需要自己处理更多细节。
高级别（High-Level）：基于 Reader 和 Writer 类型，提供自动重连、Offset 管理、超时控制等内置功能。类比自动挡——开箱即用，适合大多数场景。

学完本章后你会对这两个层级有更清晰的理解。

低级别 API：Conn 连接

Conn 类型是 kafka-go 的核心基础，封装了到 Kafka Broker 的原始 TCP 连接，提供低级别的读写接口。

生产者示例

// /producer/producer.go

topic := "my-topic"
partition := 0

// 直接连接到目标 Topic/Partition 的 Leader Broker
conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
if err != nil {
    log.Fatal("failed to dial leader:", err)
}
defer conn.Close()

// 设置写入超时
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))

// 批量写入消息
_, err = conn.WriteMessages(
    kafka.Message{Value: []byte("hi")},
    kafka.Message{Value: []byte("hello")},
    kafka.Message{Value: []byte("world")},
)
if err != nil {
    log.Fatal("failed to write messages:", err)
}

消费者示例

// /consumer/consumer.go

topic := "my-topic"
partition := 0

conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
if err != nil {
    log.Fatal("failed to dial leader:", err)
}
defer conn.Close()

// 设置读取超时
conn.SetReadDeadline(time.Now().Add(10 * time.Second))

// 批量读取：最小 10KB，最大 1MB
batch := conn.ReadBatch(10e3, 1e6)
defer batch.Close()

b := make([]byte, 10e3) // 每条消息最大 10KB
for {
    n, err := batch.Read(b)
    if err != nil {
        break // 读取完毕或出错
    }
    fmt.Println(string(b[:n]))
}

创建 Topic

方式一：自动创建（默认）

Kafka 默认配置 auto.create.topics.enable=true。这意味着当你第一次向一个不存在的 Topic 发送消息或读取消息时，Kafka 会自动创建该 Topic。

// 当 auto.create.topics.enable='true' 时，DialLeader 会自动创建不存在的 Topic
conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", "my-topic", 0)
if err != nil {
    panic(err.Error())
}

方式二：手动创建

如果你需要精确控制 Topic 的配置（分区数、副本数等），或者已经将 auto.create.topics.enable 设为 false，则需要手动创建。

在我使用的 3.9.2 版本中，auto.create.topics.enable 默认为 true，在 config/server.properties 中搜索不到该项。如需禁用，直接在文件中添加：
# 禁用自动创建 Topic
auto.create.topics.enable=false

手动创建 Topic 的代码：

topic := "my-topic"

// 1. 先连接到任意一个 Broker
conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    panic(err.Error())
}
defer conn.Close()

// 2. 获取集群控制器（Controller）的地址
controller, err := conn.Controller()
if err != nil {
    panic(err.Error())
}

// 3. 连接到 Controller
controllerConn, err := kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
if err != nil {
    panic(err.Error())
}
defer controllerConn.Close()

// 4. 定义 Topic 配置
topicConfigs := []kafka.TopicConfig{
    {
        Topic:             topic, // Topic 名称
        NumPartitions:     1,     // 分区数
        ReplicationFactor: 1,     // 副本数
    },
}

// 5. 发送创建请求
err = controllerConn.CreateTopics(topicConfigs...)
if err != nil {
    panic(err.Error())
}

通过非 Leader 连接间接连接到 Leader

有时你可能已经连接到了某个 Broker，但不一定是目标分区的 Leader。此时可以先向当前 Broker 询问 Leader 的地址，再建立到 Leader 的连接：

// 1. 连接到任意 Broker（非 Leader 也可以）
conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    panic(err.Error())
}
defer conn.Close()

// 2. 获取集群控制器信息
controller, err := conn.Controller()
if err != nil {
    panic(err.Error())
}

// 3. 连接到真正的 Leader
connLeader, err := kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
if err != nil {
    panic(err.Error())
}
defer connLeader.Close()

// 现在可以使用 connLeader 进行读写操作

列出已有 Topic

conn, err := kafka.Dial("tcp", "localhost:9092")
if err != nil {
    panic(err.Error())
}
defer conn.Close()

// 读取所有分区信息
partitions, err := conn.ReadPartitions()
if err != nil {
    panic(err.Error())
}

// 去重提取 Topic 名称
m := map[string]struct{}{}
for _, p := range partitions {
    m[p.Topic] = struct{}{}
}
for k := range m {
    fmt.Println(k)
}

Conn 是低级别 API 的基础，理解它有助于你更好地掌握 Reader 和 Writer 的工作原理。

高级别 API：Reader

Reader 是 kafka-go 提供的高级抽象，旨在简化从 Kafka 消费消息的常见场景。它自动处理以下问题：

自动重连：连接断开后自动恢复
Offset 管理：自动跟踪消费进度
超时控制：支持通过 Go 的 context 实现异步取消和超时

Important

进程退出时必须调用 Close() 关闭 Reader！ Kafka Server 需要客户端优雅断开连接才能停止向其推送消息。如果进程被 SIGINT（Ctrl+C）或 SIGTERM（docker stop / K8s 重启）直接终止而没有调用 Close()，同一个 Topic 的新消费者在连接时可能会出现延迟。建议使用 signal.Notify 捕获系统信号来执行优雅关闭。

基础消费示例

func main() {
    // 创建一个从 topic-A 分区 0、偏移量 42 开始消费的 Reader
    r := kafka.NewReader(kafka.ReaderConfig{
        Brokers:   []string{"localhost:9092"},
        Topic:     "topic-A",
        GroupID:   "order-group",
        Partition: 0,
        MaxBytes:  10e6, // 单次拉取最大 10MB
    })

    // 设置了 GroupID 后不能手动设置 Offset
    // r.SetOffset(42) // 会报错！

    // 注册信号处理，实现优雅关闭
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 消费循环
    go func() {
        for {
            m, err := r.ReadMessage(ctx)
            if err != nil {
                if ctx.Err() != nil {
                    return // 正常取消
                }
                log.Printf("读取错误: %v", err)
                continue
            }
            fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
        }
    }()

    // 等待退出信号
    <-sig
    fmt.Println("收到退出信号，正在优雅关闭...")

    cancel()
    if err := r.Close(); err != nil {
        log.Fatal("关闭 Reader 失败: ", err)
    }
    fmt.Println("程序正常退出")
}

消费者组

kafka-go 完整支持 Kafka 消费组（Consumer Group），包括由 Broker 管理的 Offset。只需在 ReaderConfig 中指定 GroupID 即可启用。

使用消费者组时，ReadMessage 会自动提交 Offset。

注意：设置 GroupID 后存在以下限制：

限制说明

(*Reader).SetOffset 会返回错误 Offset 由 Broker 统一管理，不允许手动设置

(*Reader).Offset 始终返回 -1 同上

(*Reader).Lag 始终返回 -1 消费延迟信息不可用

(*Reader).ReadLag 会返回错误同上

(*Reader).Stats 返回的分区数为 -1 统计信息受限

限制	说明
`(*Reader).SetOffset` 会返回错误	Offset 由 Broker 统一管理，不允许手动设置
`(*Reader).Offset` 始终返回 `-1`	同上
`(*Reader).Lag` 始终返回 `-1`	消费延迟信息不可用
`(*Reader).ReadLag` 会返回错误	同上
`(*Reader).Stats` 返回的分区数为 `-1`	统计信息受限

显式提交（手动确认）

如果你需要手动控制 Offset 的提交时机（例如确保消息处理成功后再提交），可以使用 FetchMessage + CommitMessages 替代 ReadMessage：

ctx := context.Background()
for {
    // FetchMessage 获取消息但不自动提交 Offset
    m, err := r.FetchMessage(ctx)
    if err != nil {
        break
    }
    fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n",
        m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))

    // 处理完消息后手动提交 Offset
    if err := r.CommitMessages(ctx, m); err != nil {
        log.Fatal("failed to commit messages:", err)
    }
}

提交规则：在消费者组中提交消息时，对于某个 Topic/Partition，Offset 最大的那条消息将决定该分区的提交位置。例如，当提交 Offset 为 3 的消息时，Offset 1 和 2 的消息也会被一并标记为已消费——这类似于计算机网络中的累积确认机制。

自动定期提交

默认情况下，CommitMessages 会同步提交 Offset 到 Kafka。为了提升性能，可以通过 CommitInterval 配置让 Offset 定期批量提交：

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:        []string{"localhost:9092", "localhost:9093", "localhost:9094"},
    GroupID:        "consumer-group-id",
    Topic:          "topic-A",
    MaxBytes:       10e6,          // 单次拉取最大 10MB
    CommitInterval: time.Second,   // 每秒自动提交一次
})

高级别 API：Writer

Writer 是对 Conn 的高级封装，用于向 Kafka 发送消息。相比低级别的 Conn，Writer 提供了：

出错时自动重试并重新建立连接
可灵活配置消息在各分区之间的分配策略（Balancer）
支持同步/异步写入
利用 context 实现异步取消
关闭时自动排空待处理消息，确保平滑关闭
支持自动创建不存在的 Topic（v0.4.30 之前为默认行为，之后需显式开启）

基础写入示例

// 创建一个 Writer，使用 LeastBytes 策略将消息分配到负载最小的分区
w := &kafka.Writer{
    Addr:     kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
    Topic:    "topic-A",
    Balancer: &kafka.LeastBytes{},
}
defer w.Close()

// 批量发送多条消息
err := w.WriteMessages(context.Background(),
    kafka.Message{
        Key:   []byte("Key-A"),
        Value: []byte("Hello World!"),
    },
    kafka.Message{
        Key:   []byte("Key-B"),
        Value: []byte("One!"),
    },
    kafka.Message{
        Key:   []byte("Key-C"),
        Value: []byte("Two!"),
    },
)
if err != nil {
    log.Fatal("failed to write messages:", err)
}

消息分区策略（Balancer）

kafka-go 提供多种分区分配策略：

策略	说明
`LeastBytes`	将消息发送到当前数据量最少的那个分区
`Hash`	根据消息的 Key 进行 Hash 取模（相同 Key → 同一分区，保证顺序）
`CRC32Balancer`	基于 CRC32 的一致性 Hash
`Murmur2Balancer`	基于 Murmur2 的一致性 Hash（Kafka 默认策略）
`RoundRobin`	轮询分配

自动创建不存在的 Topic

w := &kafka.Writer{
    Addr:                   kafka.TCP("localhost:9092"),
    Topic:                  "topic-A",
    AllowAutoTopicCreation: true, // 在 Topic 不存在时自动创建
}

messages := []kafka.Message{
    {Key: []byte("Key-A"), Value: []byte("Hello World!")},
    {Key: []byte("Key-B"), Value: []byte("One!")},
    {Key: []byte("Key-C"), Value: []byte("Two!")},
}

const retries = 3
var err error
for i := 0; i < retries; i++ {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    err = w.WriteMessages(ctx, messages...)
    // Leader 选举中或超时，等待后重试
    if errors.Is(err, kafka.LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {
        time.Sleep(time.Millisecond * 250)
        continue
    }
    if err != nil {
        log.Fatalf("unexpected error %v", err)
    }
    break
}

同时向多个 Topic 写入

通常 Writer.Topic 用于初始化单一 Topic 的 Writer。但你也可以在每条消息中单独指定 Message.Topic 来实现多 Topic 写入：

w := &kafka.Writer{
    Addr:     kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
    // 注意：这里没有设置 Topic，需要在每条消息中单独指定
    Balancer: &kafka.LeastBytes{},
}

err := w.WriteMessages(context.Background(),
    kafka.Message{
        Topic: "topic-A", // 每一条消息都指定目标 Topic
        Key:   []byte("Key-A"),
        Value: []byte("Hello World!"),
    },
    kafka.Message{
        Topic: "topic-B",
        Key:   []byte("Key-B"),
        Value: []byte("One!"),
    },
    kafka.Message{
        Topic: "topic-C",
        Key:   []byte("Key-C"),
        Value: []byte("Two!"),
    },
)
if err != nil {
    log.Fatal("failed to write messages:", err)
}

Warning

这两种模式是互斥的：

如果设置了 Writer.Topic，则不能在消息中再指定 Message.Topic
反之，如果没设置 Writer.Topic，则每条消息都必须指定 Message.Topic
Writer 检测到冲突时会直接返回错误

参考资料

kafka-go GitHub 仓库

命令行操作生产者