Kafka消息流转的挑战与对策：消息丢失与重复消费问题

    消息丢失和重复消费时分布式系统重的常见问题，如果处理不好会对业务造成很大的影响。比如用户下单是通过消息队列处理的，对于用户的订单来说，消息丢失会造成用户下单丢失，影响售卖，如果重复消费，可能会生成多个订单，多卖给了用户货物，影响也很大。

    所以，消息丢失和重复消费问题对于保证系统健壮性和业务正确性至关重要。需要了解问题出现的原因及各种解决方案。

一、消息丢失

    首先来确定什么地方可能会出现消息丢失的问题。生产者将消息发送到kafka broker中，然后消费者在来消费相应的消息。在这些环节中似乎都有可能出现消息丢失，我们来逐一分析。

1.1 生产者造成的消息丢失

生产者配置不当

    生产者发送消息主要有三种模式：发后即忘(fire-and-forget)、同步(sync)、异步(async)。

    发后既往，即ack=0，生产者只管往kafka中发送消息，但是消息有没有正确到达生产者是不知道的，这种方式吞吐量最高，但是很容易造成消息丢失。

网络故障

    生产者在将消息发往broker时，出现了网络故障，导致消息没有发送成功。这种方式通常会加入重试机制，可有效避免网络问题带来的影响。如果网络长时间不可用，那就不是消息丢失的问题了，甚至影响业务的正常运转。

生产者意外关闭

    如果生产者进程崩溃或被非正常关闭，在未完成消息发送操作之前，部分缓存中的消息可能会丢失。

    还记得以前的文章(深入剖析Kafka生产者：揭秘消息从发送到落地的全过程-CSDN博客)提到的消息发送流程吗?

    生产者客户端有两个线程协调运行，这两个线程分别为主线程和Sender线程（发送线程）。

    主线程中由KafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器(RecordAccumulator，也称为消息收集器)中。相当于将消息发送到了缓存中，如果生产者意外宕机，那缓存中的消息就会丢失，造成消息丢失。

    可以设置合理的缓存区大小，并且要尽量避免生产者服务器意外宕机的情况。

1.2 broker端故障

刷盘机制

    异步批量刷盘：Kafka Broker默认使用异步刷盘机制，先将消息存储在PageCache中再进行异步写入磁盘。如果在刷盘前Broker发生故障，这部分消息可能丢失。

    可以配置成同步刷盘策略，虽在会影响性能，但在极端情况下可以保证消息不丢失。需要根据自己的业务场景进行评估，确定一个合理的方案。

副本不完整

    当ISR集合中的副本数量不足或发生同步问题时，可能导致消息无法正确复制到其他副本上。如果生产者ack=1，消息发到leader中就认为成功了，有可能会出现消息还没同步到其他副本，leader分区出现了问题，导致消息消息丢失。

    所以要根据业务场景设置合理的ack值，减少因副本导致的消息丢失。

1.3 消费者端引起的消息丢失

自动提交偏移量

    Kafka消费者默认使用自动提交偏移量的功能，当然这个默认的自动提交不是每消费一条消息就提交一次，而是定期提交，这个定期的周期由客户端参数auto.commit.interval.ms配置，默认5S，此参数生效的前提是enable.auto.commit参数为true。

    自动提交虽然简单，但可能会造成消息丢失，比如消费者刚拉取了一批消息，然后刚好到达了提交位移的时间，刚才的消息位移就提交了，但是消费者此时出现了故障，消息还未来得及处理，这样消费者重启后就会出现消息丢失。

    在使用中尽量使用手动位移提交的方式。

手动提交处理不当

    手动提交位移如果处理不当，也会造成消息丢失，比如消费者拉取消息后，手动提交了位移，然后将消息交由其他线程来异步处理，但是处理过程出现了一场，这种情况就会出现消息丢失。

    每次提交消息都要确保消息正常的处理完毕了，否则会造成消息丢失。

消息积压造成的消息丢失

    如果生产者的发送速率远远大于消费者的出率速率，那Kafka就会产生积压，我们知道kafka是将消息保存到日志文件中的，是有留存时间的，默认是7天，倘若你的消费者程序足够慢，慢到它要消费的数据快被Kafka删除了，就会造成消费丢失。

    要监控消息积压情况并引入预警机制，如果产生了积压要及时处理，通过处理积压来解决此种原因引起的消息积压。

位移管理策略

    如果消费者设置的位移管理策略为“latest”，那么在消费者启动或者从故障中恢复时，它会直接从最新的消息开始消费，这将丢失从上次消费点到最新消息之间的所有消息。

    我们要根据业务需求，选择合理的位移管理策略。

二、重复消费

    重复消费可能会引起数据一致性问题，浪费系统计算、存储资源，甚至会影响用户体验，造成业务出错，所以要尽量避免重复消费。

2.1 生产者造成的重复消费

    生产者造成的重复消费的问题并不直观，如果生产者发送消息时存在失败重试等场景，可能会造成生成者发送重复的消息，这样就会间接的造成重复消费的问题。

    针对这种情况，消费者要做好消息的幂等处理，重复的消息只消费一次。

2.2 消费者造成的重复消费

消费者故障与重启

    当消费者在处理完消息后尚未提交偏移量就崩溃或被人为关闭，重启后会从上一次已知的偏移量继续消费，这可能导致已经处理过的消息再次被消费。

消费组再均衡

    当消费者群组中的某个消费者离开或者新加入时，整个消费者群体会触发重新平衡操作，分配新的分区给各个消费者。在此过程中，如果旧的消费者正在处理的消息未提交偏移量，则这些消息可能在新的消费者接管分区后被重新消费。

自动位移提交

    如上文所述，消费者拉取消息后，如果已经消费了一部分消息，这时位移还未提交，但是此时消费者出现了故障，消费者重启后会重新拉取消息，造成重复消费。

网络问题

    在网络不稳定的情况下，可能会发生偏移量提交请求丢失的情况，使得消费者认为某条消息未处理而进行重复消费。

    以上问题不管如何处理，都需要引入幂等机制，已经处理过的消息不在重复处理，这时终极解决方案，否则消息就有可能重复消费，对业务造成影响。

三、总结

    这一节关于kafka消息中间件出现重复消费和消息丢失的场景和原因进行了分析，你学会了吗？