使用Kafka时一定要注意防止消费速度过慢触发rebalance而导致的重复消费

本文主要是介绍使用Kafka时一定要注意防止消费速度过慢触发rebalance而导致的重复消费，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在Java应用中，我们往往会使用spring-kafka组件简单的设置一下group_id, topic就开始消费消息了，其实这样会埋下巨大的安全隐患，即当消费速度过慢时有可能会触发rebalance, 这批消息被分配到另一个消费者，然后新的消费者还会消费过慢，再次rebalance, 这样一直恶性循环下去。发生这种情况最明显的标志就是日志里能看到CommitFailedException异常，然后还会带上下面一段话：

Commit cannot be completed since the group has already rebalanced and
assigned the partitions to another member. This means that the time
between subsequent calls to poll() was longer than the configured
max.poll.interval.ms, which typically implies that the poll loop is
spending too much time message processing. You can address this either
by increasing the session timeout or by reducing the maximum size of
batches returned in poll() with max.poll.records.

其实这段话已经很走心了，kafka的开发者已经预料到了这可能是个很容易出现的问题，所以连解决方案都给你列出来了。这里我们需要明确一下，在Kafka 0.10.1.0以后的版本中，影响rebalance触发的参数有三个，说明如下：

session.timeout.ms
这个参数定义了当broker多久没有收到consumer的心跳请求后就触发rebalance，默认值是10s。在0.10.1.0之前的版本中，由于心跳请求是在poll()拉取消息的方法中执行的，因此如果当前批次处理消息耗时太长，就会导致consumer没有机会按时发送心跳，broker认为消费者已死，触发rebalance。在0.10.1.0或更新的版本中解决了这个问题，心跳请求会在单独的线程中发送，因此就不会出现因为消息处理过长而发不出心跳的问题了。
max.poll.interval.ms
这个参数定义了两次poll()之间的最大间隔，默认值为5分钟。如果超过这个间隔同样会触发rebalance。在多数情况下这个参数是导致rebalance消息重复的关键，即业务处理消息耗时太长。有人可能会疑惑，如果5分钟都没处理完消息那肯定时出了问题，其实不然。能否在5min内处理完还取决于你每次拉取了多少条消息，如果一次拿到了成千上万条的话，5min就够呛了。
max.poll.records
这个参数定义了poll()方法最多可以返回多少条消息，默认值为500。注意这里的用词是"最多"，也就是说如果在拉取消息的时候新消息不足500条，那有多少返回多少；如果超过500条，就只返回500。这个默认值是比较坑人的，如果你的消息处理逻辑比较重，比如需要查数据库，调用接口，甚至是复杂计算，那么你很难保证能够在5min内处理完500条消息，也就是说，如果上游真的突然大爆发生产了成千上万条消息，而平摊到每个消费者身上的消息达到了500的又无法按时消费完成的话就会触发rebalance, 然后这批消息会被分配到另一个消费者中，还是会处理不完，又会触发rebalance, 这样这批消息就永远也处理不完，而且一直在重复处理。