4. Go连接Kafka：使用Sarama库初始化Producer与Consumer，配置连接参数

在Go语言生态中，Sarama 是连接Apache Kafka的事实标准库。它提供了对Kafka 0.8+协议的完整实现，支持同步/异步生产者、消费者组、分区再均衡等核心特性。本章将深入讲解如何使用Sarama初始化Producer与Consumer，并解析关键连接参数的配置策略。

4.1 Sarama库概述与版本选择

Sarama由Shopify开源维护，是目前Go社区最广泛使用的Kafka客户端。其核心模块包括：

4.2 初始化Producer：同步与异步模式

Sarama的Producer提供两种发送模式：同步（SyncProducer） 和 异步（AsyncProducer）。同步模式每条消息发送后等待确认，适合对可靠性要求高的场景；异步模式通过通道批量发送，吞吐量更高。

4.2.1 同步Producer配置与初始化

package main

import (
    "fmt"
    "log"

    "github.com/IBM/sarama"
)

func main() {
    // 1. 创建配置对象
    config := sarama.NewConfig()

    // 2. 设置Producer关键参数
    config.Producer.Return.Successes = true   // 同步模式必须开启
    config.Producer.Return.Errors = true      // 错误返回
    config.Producer.RequiredAcks = sarama.WaitForAll        // 等待所有ISR副本确认
    config.Producer.Partitioner = sarama.NewHashPartitioner // 分区策略：按Key哈希
    config.Producer.Idempotent = true         // 启用幂等性（需Kafka 0.11+）
    config.Net.MaxOpenRequests = 1            // 幂等性要求最大并发请求数为1

    // 3. 创建同步Producer
    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalf("创建同步Producer失败: %v", err)
    }
    defer producer.Close()

    // 4. 发送消息
    msg := &sarama.ProducerMessage{
        Topic: "orders",
        Key:   sarama.StringEncoder("order-001"),
        Value: sarama.StringEncoder(`{"order_id": "001", "amount": 99.9}`),
    }

    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        log.Fatalf("发送消息失败: %v", err)
    }

    fmt.Printf("消息发送成功: 分区=%d, 偏移量=%d\n", partition, offset)
}

4.2.2 异步Producer配置与初始化

// 异步模式配置
config := sarama.NewConfig()
config.Producer.Return.Successes = true  // 异步模式建议开启，用于获取发送结果
config.Producer.Return.Errors = true
config.Producer.RequiredAcks = sarama.WaitForLocal  // 仅等待Leader确认，提高吞吐
config.Producer.Compression = sarama.CompressionSnappy // 启用Snappy压缩
config.Producer.Flush.Frequency = 500 * time.Millisecond // 批量刷新间隔

// 创建异步Producer
asyncProducer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)
if err != nil {
    log.Fatalf("创建异步Producer失败: %v", err)
}
defer asyncProducer.Close()

// 启动goroutine处理发送结果
go func() {
    for success := range asyncProducer.Successes() {
        fmt.Printf("消息发送成功: 分区=%d, 偏移量=%d\n", success.Partition, success.Offset)
    }
}()

go func() {
    for err := range asyncProducer.Errors() {
        log.Printf("消息发送失败: %v\n", err.Err)
    }
}()

// 发送消息（非阻塞）
asyncProducer.Input() <- &sarama.ProducerMessage{
    Topic: "orders",
    Key:   sarama.StringEncoder("order-002"),
    Value: sarama.StringEncoder(`{"order_id": "002", "amount": 199.0}`),
}

4.3 初始化Consumer与消费者组

Sarama支持两种消费方式：简单消费者（Consumer） 手动管理分区分配，以及 消费者组（ConsumerGroup） 自动协调分区再均衡。现代Go应用几乎全部使用消费者组模式。

4.3.1 消费者组配置与初始化

// 消费者组配置
config := sarama.NewConfig()
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange // 分区分配策略
config.Consumer.Offsets.Initial = sarama.OffsetOldest  // 从最早偏移量开始消费
config.Consumer.Offsets.AutoCommit.Enable = true       // 自动提交偏移量
config.Consumer.Offsets.AutoCommit.Interval = 5 * time.Second // 自动提交间隔
config.Consumer.Return.Errors = true

// 创建消费者组
group, err := sarama.NewConsumerGroup(
    []string{"localhost:9092"},
    "order-service-group",  // 消费者组ID
    config,
)
if err != nil {
    log.Fatalf("创建消费者组失败: %v", err)
}
defer group.Close()

// 实现 ConsumerGroupHandler 接口
type OrderHandler struct{}

func (h *OrderHandler) Setup(session sarama.ConsumerGroupSession) error {
    fmt.Printf("消费者组再均衡完成，分配分区: %v\n", session.Claims())
    return nil
}

func (h *OrderHandler) Cleanup(session sarama.ConsumerGroupSession) error {
    fmt.Println("消费者组再均衡开始，清理资源")
    return nil
}

func (h *OrderHandler) ConsumeClaim(
    session sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim,
) error {
    for msg := range claim.Messages() {
        fmt.Printf("收到消息: 分区=%d, 偏移量=%d, 值=%s\n",
            msg.Partition, msg.Offset, string(msg.Value))
        session.MarkMessage(msg, "") // 标记消息已处理
    }
    return nil
}

// 启动消费循环
handler := &OrderHandler{}
for {
    if err := group.Consume(context.Background(), []string{"orders"}, handler); err != nil {
        log.Printf("消费循环异常: %v", err)
    }
}

4.4 核心连接参数配置详解

Sarama的 Config 结构体包含数十个配置项，以下为生产环境最关键的参数分组：

4.5 连接参数最佳实践

4.5.1 生产环境配置模板

func NewSaramaConfig() *sarama.Config {
    config := sarama.NewConfig()

    // 网络层
    config.Net.MaxOpenRequests = 1
    config.Net.DialTimeout = 10 * time.Second
    config.Net.ReadTimeout = 30 * time.Second
    config.Net.WriteTimeout = 30 * time.Second

    // Producer
    config.Producer.Return.Successes = true
    config.Producer.Return.Errors = true
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Compression = sarama.CompressionSnappy
    config.Producer.Idempotent = true
    config.Producer.Flush.Frequency = 500 * time.Millisecond
    config.Producer.Flush.Messages = 1000

    // Consumer
    config.Consumer.Return.Errors = true
    config.Consumer.Offsets.Initial = sarama.OffsetOldest
    config.Consumer.Offsets.AutoCommit.Enable = true
    config.Consumer.Offsets.AutoCommit.Interval = 5 * time.Second
    config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky
    config.Consumer.Group.Session.Timeout = 30 * time.Second
    config.Consumer.Group.Heartbeat.Interval = 3 * time.Second

    // 元数据
    config.Metadata.Retry.Max = 3
    config.Metadata.RefreshFrequency = 10 * time.Minute

    // 版本
    config.Version = sarama.V2_8_0_0

    return config
}

4.6 连接健康检查与重连机制

Sarama内置了自动重连机制，但生产环境仍需主动监控连接状态：

• Producer健康检查：定期调用 producer.SendMessage 发送心跳消息，超时则重建Producer实例。
• 消费者组健康检查：监听 group.Errors() 通道，捕获再均衡失败或连接断开错误。
• 指标暴露：通过 sarama.MetricProvider 接口集成Prometheus，监控连接数、请求延迟、错误率。