Go 用 sarama 接入 Kafka：生产者、消费者组与分区实战

场景

Kafka 适合做削峰填谷和事件驱动。Go 侧最常用的客户端是 IBM 维护的 sarama。本文用订单事件为例，演示如何发消息、如何以消费者组方式消费并按分区并行处理。

生产者：发一条订单事件

package main

import (
	"log"

	"github.com/IBM/sarama"
)

func main() {
	brokers := []string{"localhost:9092"}
	cfg := sarama.NewConfig()
	cfg.Producer.Return.Successes = true
	cfg.Producer.Partitioner = sarama.NewRandomPartitioner

	producer, err := sarama.NewSyncProducer(brokers, cfg)
	if err != nil {
		log.Fatal(err)
	}
	defer producer.Close()

	msg := &sarama.ProducerMessage{
		Topic: "order-events",
		Key:   sarama.StringEncoder("order-1001"),
		Value: sarama.StringEncoder(`{"id":"order-1001","amount":99}`),
	}

	partition, offset, err := producer.SendMessage(msg)
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("sent to partition %d offset %d", partition, offset)
}

消费者组：批量处理事件

type handler struct{}

func (h *handler) Setup(sarama.ConsumerGroupSession) error   { return nil }
func (h *handler) Cleanup(sarama.ConsumerGroupSession) error { return nil }

func (h *handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for msg := range claim.Messages() {
		log.Printf("topic=%s partition=%d value=%s", msg.Topic, msg.Partition, string(msg.Value))
		sess.MarkMessage(msg, "")
	}
	return nil
}

func consume() {
	cfg := sarama.NewConfig()
	cfg.Consumer.Offsets.Initial = sarama.OffsetOldest
	group, _ := sarama.NewConsumerGroup([]string{"localhost:9092"}, "order-group", cfg)
	for {
		_ = group.Consume(context.Background(), []string{"order-events"}, &handler{})
	}
}

分区与消费者组的工程要点

• 同一个 group 内，一个分区只会被一个消费者实例消费，实例数不要超过分区数，否则有实例空转。
• 用稳定业务键（如 orderId）做 Key，保证同一订单的事件落到同一分区，顺序得以保留。
• MarkMessage 才提交位移，处理失败先不提交，重启可从上次位置重放。
• 生产环境务必设置 RequiredAcks 和重试，避免消息静默丢失。