1. 出问题现象
    最近项目使用到了kafka,别的系统作为生产者,我们系统作为消费者,但是经常出现消费者消费一段时间就不消费了,根本就触发不了kafkaListener的拉取动作。换一个消费者组,从最新的位置消费又可以消费的到,但是消费一段时间就又消费不到。查看服务端没任何报错,客户端也没有任何报错。
  2. 排查思路
    原因一:消费速度太慢,导致数据积压太多,所以无法消费(现在想想根本不可能)???
    由原来的1条1条拉取,改成批量消费,还是不行。改成多线程消费还是不行。
    原因二:没有给多个消费线程添加id(这个是在容器中,便于区分)
@kafkaListenner(topic="test",groupId="test11",containerFactory="kafkaLinstenerContainerFactory",id="testThead")

加了没用
原因三:心跳、会话超时时间有问题???
查阅官网资料,将参数调整为:

enable.auto.commit=true
auto.commiit.interval=100
session.timeout.ms=18000
request.timeout.ms=18000
##官方建议会话超时时间是心跳间隔的3倍
heartbeat.interval=6000

改了参数还是不行。。。
最终原因:至此不怀疑自己写的代码有问题了,开始在服务端找问题,各种百度尝试也无果。求助优秀的同时,排查了几天终于找到问题,简单来说就是限流了。这个博客kafka流量限制解释的很详细。
我们公司是利用界面化创建的kafka,创建的时候我们选着的是默认值,默认consumer.byte.rate=1024Byte。修改之后服务都正常了。
在此做一个记录,以后默认的东西还是要自己了解过再用,这个阻碍了我两个星期。。。。。。

Logo

大数据从业者之家,一起探索大数据的无限可能!

更多推荐