生产者消息发送流程详解（构建消息 → 序列化 → 分区路由 → 批量发送 → 错误处理），以及 kafka-go 与 Java 客户端的核心区别对照。

---

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

发送流程详解

1. 构建 kafka.Message：在 kafka-go 中，每条消息用 kafka.Message 结构体表示，包含：

   • Topic：目标 Topic 名称（如果 Writer 已指定则可省略）
   • Key：可选的 Key（[]byte），用于分区路由
   • Value：消息体（[]byte）
   • Headers：可选的消息头
   • Time：可选的时间戳

   msg := kafka.Message{
       Topic: "order-events",
       Key:   []byte("order-123"),
       Value: []byte(`{"event":"created"}`),
       Time:  time.Now(),
   }

2. 序列化：kafka-go 直接接受 []byte，序列化工作由调用方在传入前完成（如 json.Marshal）。这与 Java 客户端不同——没有内置的序列化器抽象。

3. 分区路由（Balancer）：Writer 通过 Balancer 接口决定消息发往哪个分区：

   • 如果消息指定了 Partition 字段 → 直接使用该分区
   • 如果消息有 Key → Balancer 对 Key 做 Hash 确定分区（如 Hash、Murmur2Balancer）
   • 如果 Key 为空 → 轮询分配（RoundRobin）或发往负载最小的分区（LeastBytes）

   w := &kafka.Writer{
       Addr:     kafka.TCP("localhost:9092"),
       Topic:    "order-events",
       Balancer: &kafka.Hash{}, // 相同 Key → 相同分区，保证顺序
   }

4. 时间戳：可在 kafka.Message.Time 中手动指定。如果留空，Broker 端的处理取决于 Topic 配置：

   • CreateTime：使用消息中携带的时间戳（如未指定则 Broker 取当前时间）
   • LogAppendTime：Broker 在追加消息到日志时覆盖为当前时间

5. 批量发送：Writer 内部会将消息打包成批次，按 BatchSize 或 BatchTimeout（哪个先满足就发送），由独立 goroutine 统一发往 Broker。WriteMessages 默认是同步的——调用会阻塞直到收到 Broker 确认。

6. Broker 响应与错误处理：

   • 写入成功 → WriteMessages 返回 nil
   • 写入失败 → 返回错误。Writer 内置自动重试（可通过 MaxAttempts 配置），常见错误如 kafka.LeaderNotAvailable（Leader 选举中）、context.DeadlineExceeded（超时）等需要区分处理

   err := w.WriteMessages(ctx, msg)
   if errors.Is(err, kafka.LeaderNotAvailable) {
       // Leader 选举中，等待后重试
       time.Sleep(250 * time.Millisecond)
   }

在 kafka-go 中对应 Java 客户端的核心区别

概念	Java 客户端	kafka-go
消息对象	ProducerRecord<K,V>	kafka.Message（纯 []byte）
序列化	内置 Serializer<K,V>	调用方自行序列化为 []byte
分区策略	Partitioner 接口	Balancer 接口
批量配置	batch.size / linger.ms	BatchSize / BatchTimeout（Writer 字段）
重试	retries 配置	MaxAttempts（Writer 字段，默认 3）
发送方式	send() 返回 Future<RecordMetadata>	WriteMessages() 同步阻塞，返回 error
幂等性	enable.idempotence=true	EnableIdempotence（Writer 字段，需 Kafka 0.11+）