当使用github.com/IBM/sarama库来编写Kafka消费者时，你可以按照以下步骤进行：
首先，确保你已经安装了sarama库。你可以使用go get命令来安装它：
bash复制代码 go get github.com/IBM/sarama
接下来，你可以编写一个简单的Kafka消费者程序。以下是一个示例代码：
go复制代码 package main import ( "fmt" "log" "github.com/IBM/sarama" ) func main() { // 配置Kafka集群的地址 brokers := []string{"localhost:9092"} // 创建消费者配置 config := sarama.NewConfig() config.ClientID = "my-consumer" config.Group.ID = "my-consumer-group" config.Version = sarama.V2_8_0_0 // 设置Kafka版本，根据你的Kafka集群版本进行调整 // 创建消费者 consumer, err := sarama.NewConsumer(brokers, config) if err != nil { log.Fatalf("Error creating consumer: %v", err) } defer consumer.Close() // 定义要消费的主题和分区 topic := "my-topic" partition := 0 // 获取分区偏移量 offset, err := consumer.GetOffset(topic, partition, sarama.OffsetOldest) if err != nil { log.Fatalf("Error getting offset: %v", err) } // 创建分区消费者 pc, err := consumer.ConsumePartition(topic, partition, offset) if err != nil { log.Fatalf("Error creating partition consumer: %v", err) } defer pc.Close() // 监听消息 for msg := range pc.Messages() { fmt.Printf("Consumed message at offset %d: %s\n", msg.Offset, string(msg.Value)) } }
在上面的示例中，我们首先配置了Kafka集群的地址，并创建了一个消费者配置对象。然后，我们使用sarama.NewConsumer函数创建了一个消费者实例。接下来，我们指定了要消费的主题和分区，并获取了分区的初始偏移量。然后，我们使用consumer.ConsumePartition函数创建了一个分区消费者，并监听该分区中的消息。最后，我们在一个无限循环中从pc.Messages()通道中读取消息，并打印出消息的内容和偏移量。
请注意，你需要根据你的Kafka集群的实际情况调整上述代码中的配置和主题/分区信息。此外，你还可以根据需要添加更多的错误处理和日志记录逻辑。
