为什么 Kafka 的数据不写了?

背景

服务器上有个 go 程序，用来抓取网卡 eth2 上的 dns 数据包，后写入 kafka 集群。但是每运行一小时左右后往 kafka 集群的写入量就变为零，因此针对这奇怪的现象进行了些许排查····· 程序处理流程如下图:

使用到的库是

github.com/confluentinc/confluent-kafka-go/kafka

github.com/google/gopacket

github.com/google/gopacket/layers

github.com/google/gopacket/pcap

排查流程

1.当写入归零时查看程序状态

top 查看:

cpu 消耗 0-1%

内存占用 1.4GB

strace 查看:

有 futex 超时现象

无其他系统调用

2.怀疑是否是链接 kafka 的 tcp 链接断开了

通过 ss 命令查看进程持有的 tcp 链接，状态为 establish

再通过，tcpdump 分析，发现无数据包传输

3.综合以上因素，并未分析出有效的原因。试试在程序关键步骤打印输出来查看。

1) 程序增加子协程心跳信息发现程序出问题时，子协程运行正常，说明程序整体运行正常

2) 在 kafka 生产者发送消息前打印日志发现，主程序 hang 在了这里(如下图),似乎是这里的问题

4.查看 confluent-kafka-go 库的 github

(https://github.com/confluentinc/confluent-kafka-go)

1) 从样例和说明文档中并未发现有特别需要注意的配置项

2) 从 issue 中搜索关键字 hang/block，发现了一个类似的问题

https://github.com/confluentinc/confluent-kafka-go/issues/251

维护者在回答某个提问者的问题时: 提到有个 Events 管道被装满，并且这是维护者也回答这个选项是开发层面的属性，所以没在配置文档中说明....

3)这个 Events 是什么东西呢，与我们发送数据又有什么关联呢，由于介绍文档中没有说明，只能从 producer.go 代码中(https://github.com/confluentinc/confluent-kafka-go/blob/master/kafka/producer.go)看是否能找到些说明..

producer.go 中有介绍到 Producer 方法是个异步方法，消息发送结果都保存在一个 Events()管道中

同时发现这个管道的容量配置默认为 100000

同时调试发现，我们程序中按照样例写的子协程，在处理一条该管道中的数据就停掉了

那么猜测就是这个管道容量用完导致程序 hang 住了。。

调试

根据上面维护者提到的解决方法把 go.delivery.reports 置为 false 重新编译运行，发现这次程序运行十几个小时写入 kafka 数据正常，内存占用 100MB 左右。

至此，就确认了问题所在....

改进与思考

1.总结

(1)confluent-kafka-go 库中的生产者 Api 发送数据是一个异步的过程。

(2)生产者 Api 通过一种叫做 delivery reports 的信息来描述该条消息发送结果成功与否，delivery reports 默认被发送进名字是 Events 的信道中。

(3)Events 管道的开关配置、最大容量在 NewProducer 实例化时配置，在实例调用 Produce 时可指定不同的 Events

(4)Events 管道容量耗尽时会阻塞生产者发送数据。

2.改进

服务器上的该 Go 程序的问题，可以有两种手段解决：

(1)在 NewProducer 实例化配置时设置 go.delivery.reports 为 false，从而关闭该功能。

(2)默认情况下 go.delivery.reports 是开启的，需要确保子协程能一直处理 Events 中的 delivery.reports 数据。

3.建议

(1)如果业务场景不能容忍数据丢失的情况，则需要:

a.NewProducer 实例化时配置 go.delivery.reports 开启

b.增加子协程来处理 Events 管道中的 delivery reports 数据，对于 delivery reports 显示消息发送失败的情况，需要增加重传的逻辑，确保消息被 Kafka 端成功收到。

c.程序运行中需要关注该 Events 管道的容量，避免管道容量耗尽的情况。

(2)如果业务可容忍数据丢失时，可关闭该功能，以提高程序的性能。

4.思考

最初 top 排查程序异常时，发现内存占用 1.4G，相对程序调试运行时的内存占用 100MB 相差十倍，作为一个非大量数据计算存储程序，内存不应占用这么多，起初的思路也可以沿着程序中是否有数据积压的可能去排查。