在go(sarama)连接卡夫卡，消费者无法通过话题获得信息

Question

我只想跟随一个演示，尝试在go中使用kafka。我可以通过sarama成功生成消息，但当我想消费消息时，却无法获得。

package main
import (
"fmt"
"github.com/Shopify/sarama"
)
// kafka consumer
func main() {
consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
if err != nil {
fmt.Printf("fail to start consumer, err:%vn", err)
return
}
partitionList, err := consumer.Partitions("test") 
if err != nil {
fmt.Printf("fail to get list of partition:err%vn", err)
return
}
fmt.Println(partitionList)
for partition := range partitionList { 
pc, err := consumer.ConsumePartition("test", int32(partition), sarama.OffsetNewest)
if err != nil {
fmt.Printf("failed to start consumer for partition %d,err:%vn", partition, err)
return
}
defer pc.AsyncClose()
go func(sarama.PartitionConsumer) {
for msg := range pc.Messages() {
fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
}
}(pc)
}
}

代码的返回是

[0]
-1

但实际上，我可以通过kafka控制台消费者获得信息。

Answer 1

我相信你不是在等待消息的到来。。。

以下是代码中的问题列表：

1. defer pc.AsyncClose()将在函数退出时触发，而不是在作用域退出时触发
2. goroutine毫无进展。。。没有什么能阻挡或等待结果的到来

go func(sarama.PartitionConsumer) {
for msg := range pc.Messages() {
fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
}
}(pc)

1. 没有将参数传递给goroutine。go func(sarama.PartitionConsumer) {这是唯一的类型。go func(pc sarama.PartitionConsumer) {

如果您想制作hello world示例，请删除goroutine，只检查消费者通道。