kafka-重复消费问题分析

喜欢ヅ旅行 2022-01-20 03:57 595阅读 0赞

**问题**

我们的系统在全量初始化数据时，需要大量发送消息到kafka，观察发现消费者出现重复消费，甚至出现持续消费但队列里的消息却未见减少的情况。虽然consumer的处理逻辑已保证幂等，重复消费并不影响数据准确，但大量的重复消费导致性能缓慢，急需解决。可以肯定的是consumer中做了大量工作，导致消费性能较慢，将消费逻辑改为多线程异步方式，重复消费问题立马解除，于是可以判断是消费性能引起重复消费问题。这是为什么呢？

**consumer和consumer group之间的关系是动态维护的**

下图是官网给出的一张图

从图中可以看到，consumer（如C1）订阅topic中的一个或者多个partition中的消息，一个consumer group下可以有多个consumer，一条消息只能被group中的一个consumer消费。consumer和consumer group的关系是动态维护的，并不固定，当某个consumer卡住或者挂掉时，该consumer订阅的partition会被重新分配给该group下其它consumer，用于保证服务的可用性。

为维护consumer和group之间的关系，consumer会定期向服务端的coordinator(一个负责维持客户端与服务端关系的协调者)发送心跳heartbeat，当consumer因为某种原因如死机无法在session.timeout.ms配置的时间间隔内发送heartbeat时，coordinator会认为该consumer已死，它所订阅的partition会被重新分配给同一group的其它consumer，该过程叫：rebalanced。

**consumer通过拉取方式读取message**

通过上述分析， 为重现问题，session.timeout.ms的默认配置时间为10秒，consumer配置为自动提交，并将consumer的消费时间延长到1分钟，却没有出现重复消费！原来，kafka在0.10.1之后的版本，增加了另一个概念：max.poll.interval.ms，即最大的poll时间间隔。consumer是通过拉取的方式向服务端拉取数据，当超过指定时间间隔max.poll.interval.ms没有向服务端发送poll()请求，而心跳heartbeat线程仍然在继续，会认为该consumer锁死，就会将该consumer退出group，并进行再分配。

这是一个巧妙的设计，两个配置项对应2个线程，在0.10.0之前的版本中，是没有区分这2个线程的，即超过session.timeout.ms没有发送心跳就直接rebalance。session.timeout.ms默认值是10秒，max.poll.interval.ms默认值是5分钟，我猜测（找不到官方说明），改进为2个线程的意义在于，heartbeat线程独立于consumer的消费能力，在后台运行，用于快速检查整个客户端服务是否可用（如发生宕机等情况），而poll线程与consumer消费能力挂勾，用于检查单个consumer是否可用，这样可以避免当某些consumer消费较久配置心跳时间很长的情况下，我们不必等到这么久才知道服务可能已经宕机了。

**poll()方法**

注意，poll方法与consumer挂勾，但并不是consumer消费完数据都会调用poll方法，除非你手动调用。它需要结合max.poll.records 每次poll的数据条数，并将数据缓存在本地，consumer消费完records后才会调用poll方法。因此，consumer的消费时长\*poll条数大于max.poll.interval.ms才会真正出现rebalanced。

**典型日志**

出现rebalanced时也能从日志里找到关键词

//撤销订阅该partition的group

ConsumerCoordinator:393 - Revoking previously assigned partitions \[\] for group group\_common\_test

//重新加入的group

AbstractCoordinator:407 - (Re-)joining group group\_common\_test

//自动提交失败

Auto-commit of offsets \{wmBiz\_topic\_india\_people\_articletype\_total\_dev2-2=OffsetAndMetadata\{offset

=191084, metadata=''\}, wmBiz\_topic\_india\_people\_articletype\_total\_dev2-3=OffsetAndMetadata\{offset=196003, metadata=''\}\} failed for group group\_db\_2\_es

\_people\_articletype\_total\_matrix\_dev: 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 pol

l 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.

**结论**

在consumer消费性能慢的情况下，在指定时间内无法自动提交，如果触发上述条件出现rebalanced，数据重新发送到新的consumer上消费，就会出现数据重复消费问题，如果一直都在间隔时间内无法完成消费，就会出现重复消费但offset不变的死循环。

因此，我们需要做到

1、保证consumer的消费性能足够快

2、尽量让consumer逻辑幂等

3、通过配置优化解决重复消费

4、不单单做好消息积压的监控，还要做消息消费速度的监控（前后两次offset比较）

上述分析也在本地重现问题，测试的consumer相关配置如下表格。

另外，kafka维护服务端与消费者服务可用性的处理思路值得学习，可以在很多场合使用。