生产者

生产者消息发送流程详解（构建消息 → 序列化 → 分区路由 → 批量发送 → 错误处理），franz-go 的压缩/批量/异步实战，kafka-go 与 Java 客户端的配置对照，以及 kafka-go 存量 Writer 配置全表。

客户端选型与基础用法见 03-Go客户端franz-go；可靠性参数（acks/重试/幂等生产）的机制展开在 08-可靠性与幂等消费。

在 Kafka 中，产生消息的一方称为生产者。消息的发送过程如下：

发送流程详解（主线 franz-go）

1. 构建 kgo.Record：每条消息用 kgo.Record 结构体表示（kafka-go 里对应 kafka.Message，字段几乎同名）：

   • Topic：目标 Topic（不填则取 DefaultProduceTopic 的兜底值）
   • Key：可选路由键（[]byte），决定分区
   • Value：消息体
   • Headers：[]kgo.RecordHeader{Key string; Value []byte}
   • Timestamp：零值取当前时间

   rec := &kgo.Record{
       Key:   []byte("SO20261013001"),
       Value: []byte(`{"event":"created"}`),
   } // Topic 由 DefaultProduceTopic("order-events") 兜底

2. 序列化：客户端只认 []byte，json.Marshal 等在调用方完成后传入。这与 Java 客户端不同——没有内置的 Serializer 抽象。

3. 分区路由（Partitioner）：franz-go 默认分区器是 UniformBytesPartitioner（KIP-794，Java 3.3+ 同款思路）：

   • 如果 Record.Partition 显式指定 → 直接使用该分区（配 kgo.ManualPartitioner() 才生效，见 09 章事故二）
   • 有 Key → murmur2 哈希——分区结果与 Java 客户端默认分区器一致（混部/迁移的安全牌）
   • 无 Key → 按字节累积自适应轮转，摊平且自动避开高积压分区

   kafka-go 用户对照：要 Java 兼容必须显式 Balancer: &kafka.Murmur2Balancer{}，选型表见 09 章。

4. 时间戳：Record.Timestamp 可手动指定。留空行为取决于 Topic 配置：

   • CreateTime：用消息自带时间戳（未指定则取发送时刻）
   • LogAppendTime：Broker 落盘时覆盖为服务端时间（判定基准的影响见 11 章）

5. 批量发送：Produce/ProduceSync 的记录先进客户端内存缓冲（默认上限 5 万条），由后台按 linger（默认 10ms）+ 批字节上限（默认 ProducerBatchMaxBytes ~1MB） 成批发往各分区 Leader。同步还是异步只决定"你什么时候拿结果"，不决定"是否攒批"。

6. Broker 响应与错误处理：

   • 协议错误统一是 *kerr.Error（Code + Retriable 标记，见 03 章）；不可重试错误（如 MessageTooLarge）重试无意义，直接走失败兜底
   • franz-go 对可重试错误自动重试：请求级 Retries（默认 20 次）+ 记录级无限次，整体由 RetryTimeout 封顶——默认没有超时上限的错觉要警惕，兜底链路依赖它按时失败
   • kafka-go 对应机制是 MaxAttempts（默认 10）+ WriteBackoffMin/Max，错误用 errors.Is 比对协议码常量

acks：一条配置的三档语义

• franz-go 默认就是 -1（AllISRAcks()）且幂等开启，业务代码什么都不用写；显式关闭幂等才会退回 RequiredAcks 可调的普通模式。
• kafka-go 的零值是 0（&kafka.Writer{} 直接构造时 RequiredAcks 默认 RequireNone——fire-and-forget），必须显式设置：

// kafka-go 存量：在 03 章速查的 w 基础上补齐"生产五件套"
w := &kafka.Writer{
    Addr:            kafka.TCP("localhost:9092", "localhost:9093", "localhost:9094"),
    Topic:           "order-events",
    Balancer:        &kafka.Murmur2Balancer{},
    RequiredAcks:    kafka.RequireAll,      // -1：等 ISR 全部落盘（机制详见 08）
    Compression:     kafka.Zstd,            // 批级压缩，见下文
    MaxAttempts:     10,
    WriteBackoffMin: 100 * time.Millisecond,
    WriteTimeout:    10 * time.Second,
}

注意：acks=all 只是客户端的等待姿势，真正的保护在 Broker 端 min.insync.replicas——裸 all 不加 min.insync 在 ISR 缩到只剩 Leader 时形同 acks=1。三件套配置在 08 章。

压缩与批量：吞吐的两个乘数

压缩发生在生产端攒批之后、整批一次（RecordBatch 级，Broker 原样落盘、消费端解压，存储侧机制见 11 章），所以批越大、同前缀模式重复越多、压缩率越高。

// franz-go：按偏好传多个编码，客户端与 Broker 协商取交集里第一个可用的
cl, _ := kgo.NewClient(
    /* ...03 章的 SeedBrokers 等... */
    kgo.ProducerBatchCompression(kgo.ZstdCompression(), kgo.SnappyCompression()),
    kgo.ProducerLinger(20 * time.Millisecond), // 多攒 20ms，吞吐与压缩率换一点延迟
)

选择逻辑：Zstd 默认首选（同等 CPU 压缩率最好）；延迟极敏感选 Lz4/Snappy（franz 与 kafka-go 的默认都够兼容）；Gzip CPU 最高，只在跨老版本兜底时用。kafka-go 写法：w.Compression = kafka.Zstd（v0.4 起四种编码内置，无需额外 import）。

注意：franz 默认 [Snappy, 不压] 协商——你没配压缩时它在支持 Snappy 的 Broker 上其实一直在压；而 kafka-go 默认完全不压、BatchTimeout 默认 1s。两个"反直觉默认"方向相反：franz 白送了压缩但linger只有10ms，kafka-go 低峰期每条消息白等 1 秒。迁移时这两处行为差异要心里有数。

异步写入：吞吐换复杂度

// franz-go：回调式异步，ProduceSync 的回调化形态
cl.Produce(ctx, rec, func(r *kgo.Record, err error) {
    if err != nil {
        log.Printf("投递失败 topic=%s key=%s: %v", r.Topic, r.Key, err)
        // 唯一的结果回收口：落 outbox / 告警（08 章）；这里 panic 会击穿进程
    }
})

需要把业务上下文带进回调就用闭包捕获（orderID 等），或塞进 Record.Headers 随消息走。三条纪律不分客户端都一样：异步后调用方拿不到即时错误，回调是唯一反馈通道；Close() 会等在途批次与回调，退出前必须调；顺序与去重语义由 08/09 章负责，异步只是提速开关。

kafka-go 对应写法：Async: true + Completion func([]kafka.Message, error) 回调，Message.WriterData 字段可以携带任意 Go 值随批回收（等价于 franz 的闭包捕获）。默认同步模式下 WriteMessages 阻塞整批返回，高并发小消息吞吐受限——kafka-go 想要异步就绕不开这两件套。

三客户端对照（franz-go / kafka-go / Java）

存量参考：kafka-go Writer 配置全表（v0.4.51 实测默认值）

Writer 是一个扁平结构体（NewWriter(WriterConfig) 是旧风格入口，直接 &kafka.Writer{} 即可）。各字段与源码注释一致的默认值如下：

注意：别拿 Java 客户端的心智模型套 kafka-go 默认值——Java acks 默认 1，kafka-go 结构体默认 0（不等确认）；Java linger.ms 默认 0，kafka-go BatchTimeout 默认 1s。方向完全相反的两个坑；franz-go 在这两项上都站在"生产安全"的一边。

实战建议

1. franz-go 用户把功夫花在默认值之外：acks/幂等/重试都已就位，要显式决策的是 RetryTimeout（兜底链路按时失败的前提）、压缩偏好、业务 Key 稳定性与发送失败的分类处理（08 章）。
2. kafka-go 用户任何业务 Writer 落地前先默写三行：RequiredAcks: RequireAll、MaxAttempts、WriteBackoffMin/Max——这是 kafka-go 默认值埋坑最密集的地方。
3. 一个进程一个客户端复用，别每次发消息 new 一个：连接、元数据缓存、批次、幂等序号都在客户端内部维护，频繁新建等于放弃批量与幂等能力。
4. 消息过大报错的排查顺序：先确认撞的是哪层限额——franz 本地/服务端都报 kerr.MessageTooLarge（对照 ProducerBatchMaxBytes 默认 ~1MB），kafka-go 先算 BatchBytes（默认 1MB，比 message.max.bytes 更早触发）→ Topic max.message.bytes → Broker message.max.bytes，按"生产批 < Topic < Broker fetch"顺序对齐。
5. 需要幂等/事务的链路直接用 franz-go（默认开启/原生支持）；还在 kafka-go 上手工模拟去重的，评估迁移成本时把 03 章的选型依据摆上桌。