• 中文
    • English
  • 注册
  • 查看作者
  • 线上kafka消息堆积,consumer掉线,怎么办?

    线上kafka消息堆积,所有consumer全部掉线,到底怎么回事?

    最近处理了一次线上故障,具体故障表现就是kafka某个topic消息堆积,这个topic的相关consumer全部掉线。

    整体排查过程和事后的复盘都很有意思,并且结合本次故障,对kafka使用的最佳实践有了更深刻的理解。

    好了,一起来回顾下这次线上故障吧,最佳实践总结放在最后,千万不要错过。

    1、现象

    • 线上kafka消息突然开始堆积

    • 消费者应用反馈没有收到消息(没有处理消息的日志)

    • kafka的consumer group上看没有消费者注册

    • 消费者应用和kafka集群最近一周内没有代码、配置相关变更

    2、排查过程

    服务端、客户端都没有特别的异常日志,kafka其他topic的生产和消费都是正常,所以基本可以判断是客户端消费存在问题。

    所以我们重点放在客户端排查上。

    1)arthas在线修改日志等级,输出debug

    由于客户端并没有明显异常日志,因此只能通过arthas修改应用日志等级,来寻找线索。

    果然有比较重要的发现:

    看起来是kafka-client自己主动发送消息给kafka集群,进行自我驱逐了。因此consumer都掉线了。

    2)arthas查看相关线程状态变量用arthas vmtool命令进一步看下kafka-client相关线程的状态。

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

     

    可以看到 HeartbeatThread线程状态是WAITING,Cordinator状态是UNJOINED。

    此时,结合源码看,大概推断是由于消费时间过长,导致客户端自我驱逐了。

    于是立刻尝试修改max.poll.records,减少一批拉取的消息数量,同时增大max.poll.interval.ms参数,避免由于拉取间隔时间过长导致自我驱逐。

    参数修改上线后,发现consumer确实不掉线了,但是消费一段时间后,还是就停止消费了。

    3、最终原因

    相关同学去查看了消费逻辑,发现了业务代码中的死循环,确认了最终原因。

    这里核心涉及到kafka的消费者和kafka之间的保活机制,可以简单了解一下。

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

     

    kafka-client会有一个独立线程HeartbeatThread跟kafka集群进行定时心跳,这个线程跟lisenter无关,完全独立。

    根据debug日志显示的“Sending LeaveGroup request”信息,我们可以很容易定位到自我驱逐的逻辑。

    线上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的死信队列。

    4.2 有没有办法通过什么手段快速发现死循环?

    一般来说,死循环的线程会导致CPU飙高、OOM等现象,在本次故障中,并没有相关异常表现,所以并没有联系到死循环的问题。

    那通过这次故障后,对kafka相关机制有了更深刻了解,poll间隔超时很有可能就是消费阻塞甚至死循环导致。

    所以,如果下次出现类似问题,消费者停止消费,但是kafkaListener线程还在,可以直接通过arthas的 thread id 命令查看对应线程的调用栈,看看是否有异常方法死循环调用。

    5、最佳实践

    通过此次故障,我们也可以总结几点kafka使用的最佳实践:

    • 使用消息队列进行消费时,一定需要多考虑异常情况,包括幂等、耗时处理(甚至死循环)的情况。

    • 尽量提高客户端的消费速度,消费逻辑另起线程进行处理,并最好做超时控制。

    • 减少Group订阅Topic的数量,一个Group订阅的Topic最好不要超过5个,建议一个Group只订阅一个Topic。

    • 参考以下说明调整参数值:max.poll.records:降低该参数值,建议远远小于<单个线程每秒消费的条数> * <消费线程的个数> * 的积。max.poll.interval.ms: 该值要大于 / (<单个线程每秒消费的条数> * <消费线程的个数>)的值。

     

    希望能够抛砖引玉,提供一些启发和思考。如果你有其他补充和建议,欢迎留言讨论。

    如果感觉本文对你有帮助,点赞关注支持一下,想要了解更多Java后端,大数据,算法领域最新资讯可以关注我公众号【架构师老毕】私信666还可获取更多Java后端,大数据,算法PDF+大厂最新面试题整理+视频精讲

  • 0
  • 0
  • 0
  • 99
  • 请登录之后再进行评论

    登录
  • 任务
  • 实时动态
  • 发布
  • 单栏布局 侧栏位置: