4. 基础生产者实现：使用 Sarama 同步/异步发送消息、分区策略、重试与幂等性配置

在 Go 语言生态中，Sarama 是最成熟、使用最广泛的 Kafka 客户端库。本章将围绕 Sarama 的 SyncProducer 与 AsyncProducer 两种模式，深入讲解消息发送的分区策略、重试机制以及幂等性配置，帮助你在生产环境中构建可靠、高性能的消息生产者。

4.1 Sarama 生产者核心架构

Sarama 的生产者设计遵循 Kafka 协议的分层模型，从应用层到底层网络层分为三个关键模块：

• RecordAccumulator：消息累积器，负责将待发送的消息按分区批量缓存，达到批次大小或超时阈值后触发发送。
• Sender：后台协程，从 Accumulator 拉取已就绪的批次，通过 TCP 连接发送到对应的 Broker。
• ResponseHandler：处理 Broker 返回的响应，根据结果触发回调、重试或错误上报。

4.2 同步生产者（SyncProducer）实现

SyncProducer 是 Sarama 提供的同步发送接口，每次调用 SendMessage 都会阻塞等待 Broker 确认，适合对消息发送结果有强实时性要求的场景。

4.2.1 基础配置与代码示例

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "time"

    "github.com/IBM/sarama"
)

func main() {
    // 1. 创建配置对象
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true  // 同步生产者必须开启
    config.Producer.Return.Errors = true     // 同步生产者必须开启
    config.Producer.Timeout = 5 * time.Second

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

    // 3. 构造消息
    msg := &sarama.ProducerMessage{
        Topic: "orders",
        Key:   sarama.StringEncoder("order-12345"),
        Value: sarama.StringEncoder(`{"order_id":"12345","status":"created"}`),
    }

    // 4. 同步发送
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        log.Printf("发送失败: %v", err)
        return
    }
    fmt.Printf("发送成功: 分区=%d, 偏移量=%d\n", partition, offset)
}

4.2.2 同步发送的适用场景

4.3 异步生产者（AsyncProducer）实现

AsyncProducer 采用非阻塞设计，调用 Input() 通道发送消息后立即返回，由后台协程负责批量发送和结果回调。适合高吞吐量、对延迟不敏感的场景。

4.3.1 异步生产者代码示例

func runAsyncProducer() {
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true
    config.Producer.Return.Errors = true
    config.Producer.Flush.Frequency = 500 * time.Millisecond  // 每500ms强制刷新
    config.Producer.Flush.Messages = 100                       // 累积100条后强制刷新

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

    // 启动协程处理成功/失败回调
    go func() {
        for success := range producer.Successes() {
            fmt.Printf("异步发送成功: 分区=%d, 偏移量=%d\n",
                success.Partition, success.Offset)
        }
    }()

    go func() {
        for err := range producer.Errors() {
            fmt.Printf("异步发送失败: %v (消息: %s)\n",
                err.Err, err.Msg.Value)
        }
    }()

    // 发送消息
    for i := 0; i < 1000; i++ {
        msg := &sarama.ProducerMessage{
            Topic: "orders",
            Key:   sarama.StringEncoder(fmt.Sprintf("key-%d", i)),
            Value: sarama.StringEncoder(fmt.Sprintf(`{"seq":%d}`, i)),
        }
        producer.Input() <- msg
    }

    // 等待发送完成
    time.Sleep(2 * time.Second)
}

4.4 分区策略详解

分区策略决定了消息被发送到哪个 Partition。Sarama 提供了多种内置分区器，也支持自定义实现。

4.4.1 内置分区器对比

4.4.2 自定义分区器实现

当内置分区器无法满足业务需求时，可以实现 Partitioner 接口：

type MyPartitioner struct{}

func (p *MyPartitioner) Partition(msg *sarama.ProducerMessage, numPartitions int32) (int32, error) {
    // 根据消息中的业务字段决定分区
    if msg.Key == nil {
        return 0, nil
    }
    // 示例：将 key 以 "vip_" 开头的消息路由到分区 0
    keyStr := string(msg.Key.(sarama.StringEncoder))
    if strings.HasPrefix(keyStr, "vip_") {
        return 0, nil
    }
    // 其他消息使用哈希
    hash := fnv.New32a()
    hash.Write([]byte(keyStr))
    return int32(hash.Sum32() % uint32(numPartitions)), nil
}

// 使用自定义分区器
config.Producer.Partitioner = func(topic string) sarama.Partitioner {
    return &MyPartitioner{}
}

4.5 重试机制配置

Sarama 的重试机制分为两个层级：生产者级别重试和网络级别重试。合理配置重试参数可以显著提升消息投递的可靠性。

4.5.1 核心重试参数

4.5.2 重试配置示例

config := sarama.NewConfig()
config.Producer.Retry.Max = 5
config.Producer.Retry.Backoff = 300 * time.Millisecond
config.Net.MaxOpenRequests = 10
config.Net.DialTimeout = 10 * time.Second
config.Net.ReadTimeout = 30 * time.Second

// 注意：当启用幂等性时，重试次数会被自动限制
config.Producer.Idempotent = true  // 启用幂等性

4.6 幂等性配置

Kafka 的幂等性生产者（Idempotent Producer）可以防止因重试导致的消息重复写入。Sarama 从 v1.25.0 开始支持幂等性配置。

4.6.1 幂等性原理

幂等性通过以下机制保证消息精确一次（Exactly Once）语义：

• Producer ID（PID）：每个生产者实例分配一个唯一的 PID。
• 序列号（Sequence Number）：每个分区维护一个单调递增的序列号，Broker 根据序列号去重。
• 幂等性保证范围：仅保证单个生产者会话内、单个分区的消息不重复。

4.6.2 幂等性配置要求

4.6.3 幂等性配置代码

config := sarama.NewConfig()
config.Producer.Idempotent = true
config.Producer.RequiredAcks = sarama.WaitForAll
config.Net.MaxOpenRequests = 1
config.Producer.Retry.Max = 10  // 幂等性下会被自动限制

producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
if err != nil {
    log.Fatalf("创建幂等生产者失败: %v", err)
}
defer producer.Close()

// 发送消息（幂等性保证自动生效）
msg := &sarama.ProducerMessage{
    Topic: "orders",
    Key:   sarama.StringEncoder("order-999"),
    Value: sarama.StringEncoder(`{"order_id":"999","status":"paid"}`),
}
partition, offset, err := producer.SendMessage(msg)
if err != nil {
    log.Printf("发送失败: %v", err)
    return
}
fmt.Printf("幂等发送成功: 分区=%d, 偏移量=%d\n", partition, offset)

4.7 同步 vs 异步生产者对比总结

4.8 生产环境最佳实践

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

    // 基础配置
    config.Version = sarama.V2_6_0_0
    config.ClientID = "order-service-producer"

    // 生产者核心配置
    config.Producer.Return.Successes = true
    config.Producer.Return.Errors = true
    config.Producer.RequiredAcks = sarama.WaitForLocal  // 等待 Leader 确认
    config.Producer.Compression = sarama.CompressionSnappy
    config.Producer.Flush.Frequency = 100 * time.Millisecond
    config.Producer.Flush.Messages = 500
    config.Producer.Flush.Bytes = 65536  // 64KB

    // 重试配置
    config.Producer.Retry.Max = 5
    config.Producer.Retry.Backoff = 200 * time.Millisecond

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

    // 分区策略：使用 Hash 分区器保证相同 Key 的顺序性
    config.Producer.Partitioner = sarama.NewHashPartitioner

    return config
}

通过本章的学习，你已经掌握了 Sarama 生产者的核心实现方式，包括同步/异步模式的选择、分区策略的配置、重试机制的调优以及幂等性的启用条件。在实际项目中，建议先根据业务场景选择生产者模式，再逐步调整参数以达到吞吐量与可靠性的最佳平衡。