kafka 消费者参数 Kafka heartbeat.interval.ms 与 session.timeout.ms与 max.poll.interval.ms

ゝ一世哀愁。 2023-02-28 08:50 1阅读 0赞

[Kafka heartbeat.interval.ms 与 session.timeout.ms][Kafka heartbeat.interval.ms _ session.timeout.ms]

\*\*在kafka0.10.1之后的版本中，将session.timeout.ms 和 max.poll.interval.ms 解耦了。\*\*也就是说：new KafkaConsumer对象后，在while true循环中执行consumer.poll拉取消息这个过程中，其实背后是有2个线程的，即一个kafka consumer实例包含2个线程：一个是heartbeat 线程，另一个是processing线程，processing线程可理解为调用consumer.poll方法执行消息处理逻辑的线程，而heartbeat线程是一个后台线程，对程序员是"隐藏不见"的。如果消息处理逻辑很复杂，比如说需要处理5min，那么 max.poll.interval.ms可设置成比5min大一点的值。而heartbeat 线程则和上面提到的参数 heartbeat.interval.ms有关，heartbeat线程 每隔heartbeat.interval.ms向coordinator发送一个心跳包，证明自己还活着。只要 heartbeat线程 在 session.timeout.ms 时间内 向 coordinator发送过心跳包，那么 group coordinator就认为当前的kafka consumer是活着的。

\*\*在kafka0.10.1之前，发送心跳包和消息处理逻辑这2个过程是耦合在一起的，\*\*试想：如果一条消息处理时长要5min，而session.timeout.ms=3000ms，那么等 kafka consumer处理完消息，group coordinator早就将consumer 移出group了，因为只有一个线程，在消息处理过程中就无法向group coordinator发送心跳包，超过3000ms未发送心跳包，group coordinator就将该consumer移出group了。而将二者分开，一个processing线程负责执行消息处理逻辑，一个heartbeat线程负责发送心跳包，那么：就算一条消息需要处理5min，只要底heartbeat线程在session.timeout.ms向group coordinator发送了心跳包，那consumer可以继续处理消息，而不用担心被移出group了。另一个好处是：如果consumer出了问题，那么在 session.timeout.ms内就能检测出来，而不用等到 max.poll.interval.ms 时长后才能检测出来。

作者：taj3991  
链接：https://www.jianshu.com/p/408f1de5aeaf  
来源：简书  
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

[Kafka heartbeat.interval.ms _ session.timeout.ms]: https://www.jianshu.com/p/408f1de5aeaf