线上kafka消息堆积,consumer掉线,怎么办?( 二 )


文章插图
HeartbeatThread线程在发送心跳前,会比较一下当前时间跟上次poll时间,一旦大于max.poll.interval.ms 参数,就会发起自我驱逐了 。
4、进一步思考虽然最后原因找到了,但是回顾下整个排查过程,其实并不顺利,主要有两点:

  • kafka-client对某个消息消费超时能否有明确异常?而不是只看到自我驱逐和rebalance
  • 有没有办法通过什么手段发现 消费死循环?
4.1 kafka-client对某个消息消费超时能否有明确异常?4.1.1 kafka似乎没有类似机制我们对消费逻辑进行断点,可以很容易看到整个调用链路 。
线上kafka消息堆积,consumer掉线,怎么办?

文章插图
对消费者来说,主要采用一个线程池来处理每个kafkaListener,一个listener就是一个独立线程 。
这个线程会同步处理 poll消息,然后动态代理回调用户自定义的消息消费逻辑,也就是我们在@KafkaListener中写的业务 。
线上kafka消息堆积,consumer掉线,怎么办?

文章插图
所以,从这里可以知道两件事情 。
第一点,如果业务消费逻辑很慢或者卡住了,会影响poll 。
第二点,这里没有看到直接设置消费超时的参数,其实也不太好做 。
因为这里做了超时中断,那么poll也会被中断,是在同一个线程中 。所以要么poll和消费逻辑在两个工作线程,要么中断掉当前线程后,重新起一个线程poll 。
所以从业务使用角度来说,可能的实现,还是自己设置业务超时 。比较通用的实现,可以是在消费逻辑中,用线程池处理消费逻辑,同时用Future get阻塞超时中断 。
google了一下,发现kafka 0.8 曾经有consumer.timeout.ms这个参数,但是现在的版本没有这个参数了,不知道是不是类似的作用 。
4.1.2 RocketMQ有点相关机制然后去看了下RocketMQ是否有相关实现,果然有发现 。
在RocketMQ中,可以对consumer设置consumeTimeout,这个超时就跟我们的设想有一点像了 。
consumer会启动一个异步线程池对正在消费的消息做定时做 cleanExpiredMsg() 处理 。
线上kafka消息堆积,consumer掉线,怎么办?

文章插图
注意,如果消息类型是顺序消费(orderly),这个机制就不生效 。
如果是并发消费,那么就会进行超时判断,如果超时了,就会将这条消息的信息通过sendMessageBack() 方法发回给broker进行重试 。
线上kafka消息堆积,consumer掉线,怎么办?

文章插图
如果消息重试超过一定次数,就会进入RocketMQ的死信队列 。
spring-kafka其实也有做类似的封装,可以自定义一个死信topic,做异常处理

经验总结扩展阅读