Go Kafka Sarama 消费组消费代码示例

以下是基于 Sarama 库实现 Kafka 消费组消费的示例代码：

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/Shopify/sarama"
)

func main() {
	// 创建 Kafka 消费者配置对象
	config := sarama.NewConfig()

	// 配置消费者组 ID
	config.Consumer.Group.Session.Timeout = 10 * time.Second
	config.Consumer.Group.Heartbeat.Interval = 3 * time.Second
	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	config.Consumer.Offsets.Initial = sarama.OffsetOldest

	// 创建消费者组对象
	consumerGroup, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)
	if err != nil {
		panic(err)
	}

	// 定义消费者组处理函数
	handler := &ConsumerGroupHandler{
		ready: make(chan bool),
	}

	// 启动消费者组
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			if err := consumerGroup.Consume(context.Background(), []string{"my-topic"}, handler); err != nil {
				fmt.Printf("Error from consumer: %v", err)
			}
			// 消费者组退出，等待 5 秒后重新启动
			time.Sleep(5 * time.Second)
		}
	}()

	// 等待消费者组准备就绪
	<-handler.ready

	// 监听系统信号，优雅退出
	sigterm := make(chan os.Signal, 1)
	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
	select {
	case <-sigterm:
		fmt.Println("terminating: received signal")
		consumerGroup.Close()
	case <-handler.errors:
		fmt.Println("terminating: received error")
		consumerGroup.Close()
	}

	// 等待消费者组退出完成
	wg.Wait()
}

// 定义消费者组处理函数结构体
type ConsumerGroupHandler struct {
	ready  chan bool
	errors chan error
}

// 实现消费者组处理函数接口
func (h *ConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
	close(h.ready)
	return nil
}

func (h *ConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		fmt.Printf("Message claimed: value = '%s', timestamp = %v, topic = %s, partition = %v, offset = %v\n", string(message.Value), message.Timestamp, message.Topic, message.Partition, message.Offset)
		session.MarkMessage(message, "")
	}
	return nil
}

以上代码中，我们创建了一个 Kafka 消费者组，并定义了一个消费者组处理函数结构体 ConsumerGroupHandler，实现了 sarama.ConsumerGroupHandler 接口的三个方法。其中，Setup 方法在消费者组启动后执行一次，用于做一些初始化工作；ConsumeClaim 方法在消费者组启动后一直执行，每当消费者组分配到新的消息分区时就会调用，用于消费该分区的消息；Cleanup 方法在消费者组关闭前执行一次，用于清理一些资源。

在 main 函数中，我们创建了一个消费者组对象，并指定了消费者组处理函数。然后启动了一个 goroutine，不断地消费消息，如果消费过程中出现了错误，等待 5 秒后重新启动。最后，我们监听系统信号，优雅退出消费者组。