Golang segmentio/kafka-go 消费者不工作

Question

我正在使用 segmentio/kafka-go连接到卡夫卡。

// to produce messages
topic := "my-topic"
partition := 0

conn, _ := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)

conn.SetWriteDeadline(time.Now().Add(10*time.Second))
conn.WriteMessages(
    kafka.Message{Value: []byte("one!")},
    kafka.Message{Value: []byte("two!")},
    kafka.Message{Value: []byte("three!")},
)

conn.Close()

我能够使用此代码将数据生成到我的 Kafka 服务器中。

// to consume messages
topic := "my-topic"
partition := 0

conn, _ := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)

conn.SetReadDeadline(time.Now().Add(10*time.Second))
batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max

b := make([]byte, 10e3) // 10KB max per message
for {
    _, err := batch.Read(b)
    if err != nil {
        // err -> "invalid codec"
        break
    }
    fmt.Println(string(b))
}

batch.Close()
conn.Close()

但是我无法使用上面的代码消费。我收到错误 无效的编解码器。可能是什么原因？

在相关情况下，我将最小批量大小调整为 1，以便它尝试消耗一些东西。

Answer 1

只是一个猜测:尝试添加导入以加载压缩编解码器，以防您的主题使用压缩。

导入_“github.com/segmentio/kafka-go/snappy”