kafka 重启consumer 重复消费问题

拼搏现实的明天。 2022-12-21 00:44 469阅读 0赞

kafka重复消费消息  
    1，服务端配置offset失效时间

在特殊情况下，将log.retention.hours（168小时= 7天）和offsets.retention.minutes（1440分钟= 1天）的默认值结合使用可能很危险。偏移保留应始终大于对数保留。

我们已经观察到以下情况和问题：

*  生产者更新已于两天前禁止了该主题的数据生产，该主题未删除。
 *  消费者使用了所有数据并正确分配了对Kafka的偏移量。
 *  消费者不再对该主题进行抵消提交，因为没有更多传入数据，也没有任何可确认的内容。（我们已禁用自动提交功能，我不确定启用自动提交的行为方式。）
 *  一天后：Kafka根据offsets.retention.minutes清除了太旧的偏移量。
 *  两天后：长期运行的使用者在更新后重新启动，由于该主题已被offsets.retention.minutes删除，因此未找到该主题的任何已提交偏移量，因此从一开始就开始使用它。
 *  由于log.retention.hours较长，消息仍在Kafka中，大约5天的消息被再次读取。

解决此问题的已知解决方法：

*  显式配置log.retention.hours和offsets.retention.minutes，不要使用默认值。
 *  Kafka Server端保存的offset的过期时间。默认值1440（1440分钟也就是24小时），应该调整为与log.retention.hours一致，即10080。

2，客户端配置消费者从最后一个开始记录，并且自动提交消费记录

spring:
      kafka:
        bootstrap-servers: ${KAFKA_BROKERADDRESS:${LOCAL_KAFKA_BROKERADDRESS}}
        producer:
          # 发生错误后，消息重发的次数。
          retries: 1
          #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
          batch-size: 16384
          # 设置生产者内存缓冲区的大小。
          buffer-memory: 33554432
          # 键的序列化方式
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          # 值的序列化方式
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
          # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
          # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
          acks: 1
        consumer:
          # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
          auto-commit-interval: 1S
          # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
          # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
          # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
    #      auto-offset-reset: earliest
          # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
          enable-auto-commit: false  
          # 键的反序列化方式
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          # 值的反序列化方式
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        listener:
          # 在侦听器容器中运行的线程数。
          concurrency: 5
          #listner负责ack，每调用一次，就立即commit
          ack-mode: manual_immediate
          missing-topics-fatal: false

参考：

[https://www.cnblogs.com/lshan/p/12573631.html][https_www.cnblogs.com_lshan_p_12573631.html]

[https://blog.csdn.net/usagoole/article/details/82813091][https_blog.csdn.net_usagoole_article_details_82813091]

[https_www.cnblogs.com_lshan_p_12573631.html]: https://www.cnblogs.com/lshan/p/12573631.html
[https_blog.csdn.net_usagoole_article_details_82813091]: https://blog.csdn.net/usagoole/article/details/82813091