Kafka 重启导致消费者接收损坏的消息

一分耕耘，一分收获！既然都打开这篇《Kafka 重启导致消费者接收损坏的消息》，就坚持看下去，学下去吧！本文主要会给大家讲到等等知识点，如果大家对本文有好的建议或者看到有不足之处，非常欢迎大家积极提出！在后续文章我会继续更新Golang相关的内容，希望对大家都有所帮助！

我正在运行具有复制因子 2 的 Kafka 服务，生产者要求 1 写入时确认。 在消费者方面，我们不使用消费者组，所有消费者都连接到显式配置的分区号。

我们多次观察到，当我们对 Kafka 集群进行滚动重启时，当 Kafka 服务器出现故障时，正在消费该 Kafka 服务器作为领导者的分区的消费者会收到一些损坏的消息。毫秒。

我还没有识别出损坏的模式，看起来消息的某些部分只是被随机值替换了。即使应该始终为 ascii 字符的字段也会以非 ascii 字节值结束。

在 Kafka 未重新启动的所有其他时间里，消费者永远不会收到任何这些损坏的消息，这只在 Kafka 重新启动时才会发生。

在生产者和消费者方面，我们使用segment-io Kafka客户端：https://github.com/segmentio/kafka-go

我目前还不确定这是否是 Kafka 问题，还是 Segment-io Kafka 客户端的问题。

正确答案

在生产者端，我们使用 segmentio/kafka-go 的 writer.writemessages() 将消息发布到 kafka，该方法使用上下文来异步取消操作并发布 kafka 消息：

func (w *Writer) WriteMessages(ctx context.Context, msgs ...Message) error

在发布消息后，我们会池化并重新使用给定消息中的字节切片，以减少 go gc 的负载，这通常工作得很好，除了在 .writemessages() 传递上下文的情况下在成功写入传递的消息之前被取消。

我们将超时为 2s 的上下文传递给 .writemessages()，这通常没有问题，因为在消息写入 kafka 之前，它永远不会花费 2s，除非其中一个 kafka 关闭的情况宕机，写入器需要切换到为受影响分区提供服务的其他副本。在这种情况下，上下文超时，.writemessages() 返回错误，然后我们重新使用之前传递给它的消息的字节片来构建下一批消息，但此时作者仍然将这些消息放在其内部队列中，并在我们重新使用它们之后写入它们。

我已向 segmentio/kafka-go 项目提交了 pr，以在评论中澄清这一点：

https://github.com/segmentio/kafka-go/pull/1139

以上就是《Kafka 重启导致消费者接收损坏的消息》的详细内容，更多关于的资料请关注golang学习网公众号！