在 Golang 中使用 Kafka 时，Kafka 的消费者组使用了 rebalance 机制来分配分区和重新分配分区。rebalance 是指在以下情况下发生的一种事件：
- 新的消费者加入到消费者组中；
- 消费者从消费者组中离开；
- 分区数量发生变化。
当出现以上情况时，Kafka 将会触发 rebalance 事件，以重新平衡每个消费者所负责的分区。
以下是 sarama 库实现 Kafka 消费者组 rebalance 机制的示例代码：
config := sarama.NewConfig()
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
consumer, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "test-group", config)
if err != nil {
panic(err)
}
defer consumer.Close()
handler := &myHandler{}
for {
err = consumer.Consume(context.Background(), []string{"test-topic"}, handler)
if err != nil {
panic(err)
}
}
以上代码中创建了一个新的 ConsumerGroup 实例，并将 Rebalance 策略设置为 RoundRobin 表示轮询方式。然后循环调用 Consume() 方法并传入自定义处理函数 myHandler，以实现对消息的处理。
在自定义处理函数中需要实现 sarama.ConsumerGroupHandler 接口，并重写三个方法：Setup()、Cleanup() 和 ConsumeClaim()。其中，Setup() 方法会在 consumer group 成功启动时被调用，而 Cleanup() 方法则会在 consumer group 关闭时被调用。ConsumeClaim() 方法则是实际处理消息的方法。
以下是示例代码中自定义处理函数的实现：
type myHandler struct{}
func (h *myHandler) Setup(sess sarama.ConsumerGroupSession) error {
fmt.Printf("New session setup: %v\n", sess)
return nil
}
func (h *myHandler) Cleanup(sess sarama.ConsumerGroupSession) error {
fmt.Printf("Session cleanup: %v\n", sess)
return nil
}
func (h *myHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("Message claimed: value = %s, topic = %s, partition = %d, offset = %d\n",
string(msg.Value), msg.Topic, msg.Partition, msg.Offset)
// do something with the message
sess.MarkMessage(msg, "")
}
return nil
}
以上代码中实现了 sarama.ConsumerGroupHandler 接口，并重写了三个方法：Setup()、Cleanup() 和 ConsumeClaim()。在 Setup() 方法中打印出新的 session 信息，在 Cleanup() 方法中打印出关闭 session 的信息。在 ConsumeClaim() 方法中对 claim 中的每条消息进行处理，并标记已经处理过的消息。
需要注意的是，当消费者组发生 rebalance 事件时，Kafka 会调用一次 Cleanup() 方法来释放掉当前所有分配给该消费者组的分区，然后再调用一次 Setup() 方法来分配新的分区。在这个过程中，消费者需要暂停消息处理并等待新的分区分配完成后再继续处理消息。
以上是 sarama 库实现 Kafka 消费者组 rebalance 机制的示例代码，其他库也大同小异。
