Kafka重复消费

大家好，欢迎来到IT知识分享网。

重复消费的原因

　　消息重复消费的根本原因都在于：已经消费了数据，但是offset没有成功提交。

　　其中很大一部分原因在于发生了再均衡。

　　　　1）消费者宕机、重启等。导致消息已经消费但是没有提交offset。

　　　　2）消费者使用自动提交offset，但当还没有提交的时候，有新的消费者加入或者移除，发生了rebalance。再次消费的时候，消费者会根据提交的偏移量来，于是重复消费了数据。

　　　　3）消息处理耗时，或者消费者拉取的消息量太多，处理耗时，超过了max.poll.interval.ms的配置时间，导致认为当前消费者已经死掉，触发再均衡。

重复消费的解决方案

　　由于网络问题，重复消费不可避免，因此，消费者需要实现消费幂等。

　　方案：

　　　　①：消息表

　　　　②：数据库唯一索引

　　　　③：缓存消费过的消息id

　　　　提高消费者的处理速度。例如：对消息处理中比较耗时的步骤可通过异步的方式进行处理、利用多线程处理等。在缩短单条消息消费的同时，根据实际场景可将max.poll.interval.ms值设置大一点，避免不必要的Rebalance。可根据实际消息速率适当调小max.poll.records的值。

重复消费问题：

　　每次拉取的消息记录数max.poll.records为100，poll最大拉取间隔max.poll.interval.ms为 300s，消息处理过于耗时导致时长大于了这个值，导致再均衡发生重复消费

　　解决办法：

　　　　①：减少每次拉取的消息记录数和增大poll之间的时间间隔

　　　　②：拉取到消息之后异步处理（保证成功消费）

注意：kafka新版本已经在broker中保证了接收消息的幂等性（比如2.4版本），只需在生产者加上参数 props.put(“enable.idempotence”, true) 即可，默认是false不开启。

新版kafka的broker幂等性具体实现原理：
　　kafka每次发送消息会生成PID和Sequence Number，并将这两个属性一起发送给broker，broker会将PID和Sequence Number跟消息绑定一起存起来，下次如果生产者重发相同消息，broker会检查PID和　　　　Sequence Number，如果相同不会再接收。

　　PID：每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，这个PID对用户完全是透明的。生产者如果重启则会生成新的PID。
　　Sequence Number：对于每个 PID，该 Producer 发送到每个 Partition 的数据都有对应的序列号，这些序列号是从0开始单调递增的。

常见配置

fetch.min.byte：配置Consumer一次拉取请求中能从Kafka中拉取的最小数据量，默认为1B，如果小于这个参数配置的值，就需要进行等待，直到数据量满足这个参数的配置大小。调大可以提交吞吐量，但也会造成延迟

fetch.max.bytes，一次拉取数据的最大数据量，默认为52428800B，也就是50M，但是如果设置的值过小，甚至小于每条消息的值，实际上也是能消费成功的

fetch.wait.max.ms，若是不满足fetch.min.bytes时，等待消费端请求的最长等待时间，默认是500ms

max.poll.records，单次poll调用返回的最大消息记录数，如果处理逻辑很轻量，可以适当提高该值。一次从kafka中poll出来的数据条数,max.poll.records条数据需要在在session.timeout.ms这个时间内处理完，默认值为500

consumer.poll(100) ，100 毫秒是一个超时时间，一旦拿到足够多的数据（fetch.min.bytes 参数设置），consumer.poll(100)会立即返回 ConsumerRecords<String, String> records。如果没有拿到足够多的数据，会阻塞100ms，但不会超过100ms就会返回

max.poll.interval.ms，两次拉取消息的间隔，默认5分钟；通过消费组管理消费者时，该配置指定拉取消息线程最长空闲时间，若超过这个时间间隔没有发起poll操作，则消费组认为该消费者已离开了消费组，将进行再均衡操作（将分区分配给组内其他消费者成员）

若超过这个时间则报如下异常：

org.apache.kafka.clients.consumer.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.　

　　即：无法完成提交，因为组已经重新平衡并将分区分配给另一个成员。这意味着对poll()的后续调用之间的时间比配置的max.poll.interval.ms长，这通常意味着poll循环花费了太多的时间来处理消息。

可以通过增加max.poll.interval.ms来解决这个问题，也可以通过减少在poll()中使用max.poll.records返回的批的最大大小来解决这个问题

max.partition.fetch.bytes：该属性指定了服务器从每个分区返回给消费者的最大字节数，默认为 1MB。

session.timeout.ms：消费者在被认为死亡之前可以与服务器断开连接的时间，默认是 3s，将触发再均衡操作

　　对于每一个Consumer Group，Kafka集群为其从Broker集群中选择一个Broker作为其Coordinator。Coordinator主要做两件事：

1. 维持Group成员的组成。这包括加入新的成员，检测成员的存活性，清除不再存活的成员。

2. 协调Group成员的行为。

poll机制

　　　　①：每次poll的消息处理完成之后再进行下一次poll，是同步操作

　　　　②：每次poll之前检查是否可以进行位移提交，如果可以，那么就会提交上一次轮询的位移

　　　　③：每次poll时，consumer都将尝试使用上次消费的offset作为起始offset，然后依次拉取消息

　　　　④：poll(long timeout)，timeout指等待轮询缓冲区的数据所花费的时间，单位是毫秒