Kafka数据如何保障不丢失_kafka消费不丢失

Kafka 消息丢失的场景以及解决方案

本文将由Producer->Broker->Consumer消息传递的三个参与者分别进行分析丢失消息的场景以及解决方案

Producer生产者

1. 场景
   • Kafka Producer 是异步发送消息，如果Producer 客户端使用了 producer.send(msg) 方法来发送消息，方法会立即返回，但并不能代表消息已经发送成功了。如果消息在发送的过程中发生了网络抖动或者消息体过大等情况，那么消息可能会丢失。
2. 方案

   • 使用 producer.send(msg, callback) 带有回调通知的 send 方法可以针对发送失败的消息进行重试处理,比如把失败消息存储到数据库进行记录等待重试

   • 设置 acks参数
     设置 acks = all/-1。此配置是 Producer 在确认一个请求发送完成之前需要收到的反馈信息的数量。 这个参数是为了保证发送请求的可靠性

     acks=0，producer 不会等待服务器的反馈。该消息会被立刻添加到 socket buffer 中并认为已经发送完成。在这种情况下，服务器是否收到请求是没法保证的，并且参数retries也不会生效（因为客户端无法获得失败信息）。每个记录返回的 offset 总是被设置为-1。
     acks=1，leader节点会将记录写入本地日志，并且在所有 follower 节点反馈之前就先确认成功。在这种情况下，如果 leader 节点在接收记录之后，并且在 follower 节点复制数据完成之前产生错误，则这条记录会丢失。
     acks=all，这就意味着 leader 节点会等待所有同步中的副本确认之后再确认这条记录是否发送完成。只要至少有一个同步副本存在，记录就不会丢失。这种方式是对请求传递的最有效保证。acks=-1与acks=all是等效的。

   • 设置重试参数 retries, retry.backoff.ms
     当出现网络的瞬时抖动等可重试异常，如果配置了 retries > 0 （一般设置为3），那么Producer 会进行自动重试消息发送，避免消息丢失。如果重试达到设定的次数，那么生产者则会放弃重试并返回异常。所以需要合理估算重试的时间间隔，通过retry.backoff.ms用来设定两次重试之间的时间间隔，避免无效的频繁重试。

Broker服务端

1. 场景

   • Leader Broker 宕机了，触发选举过程，集群选举了一个落后 Leader 太多的 Follower作为 Leader，那么落后的那些消息就会丢失了。
   • Kafka 为了提升性能，使用页缓存机制，将消息写入页缓存而非直接持久化至磁盘，采用了异步批量刷盘机制，也就是说，按照一定的消息量和时间间隔去刷盘，刷盘的动作由操作系统来调度的，如果刷盘之前，Broker 宕机了，重启后在页缓存的这部分消息则会丢失。

2. 方案

   • 多副本保障机制参数设置
     unclean.leader.election.enable：设置为false，拒绝从ISR集合之外的副本中选举leader副本。这样的话就不用担心选举出来的副本消息落后原leader副本太多。
     replication.factor：设置大于等于3，这样当leader副本宕机之后才有follower副本选举成leader。
     min.insync.replicas：设置为大于1，指定ISR集合中最小的副本数，保证了ISR集合中不会只有leader副本的情况出现。推荐设置成 replication.factor = min.insync.replicas + 1，如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。

   • 调整刷盘参数
     当 Kafka 服务器接收到消息后，并不直接写入磁盘，而是先写入页内存PageCache中，Kafka 服务端会根据不同设置参数，选择不同的刷盘过程，这里有两个参数控制着这个刷盘过程：
     log.flush.interval.messages：在将消息刷新到磁盘之前，在日志分区上累积的消息数量。
     log.flush.interval.ms：在刷新到磁盘之前，任何topic中的消息保留在内存中的最长时间（以毫秒为单位）。如果未设置，则使用log.flush.scheduler.interval.ms中的值。
     如果设置 log.flush.interval.messages=1，那么每次来一条消息，就会刷一次磁盘。通过这种方式，就可以降低消息丢失的概率，但也会严重影响性能。但要注意的是，Kafka 服务端返回确认之后，仅仅表示该消息已经写入到 Kafka 服务器的 PageCache 中，并不代表其已经写入磁盘了。这时候如果 Kafka 所在服务器断电或宕机，那么消息也是丢失了。

Consumer消费者

1. 场景
   • 消费者拉取了消息，（自动/手动）提交了偏移量，但消费发生异常，导致丢失此条消息。
2. 方案
   • 确保消息消费完成再提交。
     采用手动提交位移的方式设置 enable.auto.commit = false。批量消费时可以根据现实情况调整提交offset的频次。（每条消息处理完成都进行提交，会影响消速率。但等消息全部结束再进行提交会提高重复消费的可能性）
     虽然采用手动提交位移的方式可以解决消费端消息丢失的场景，但同时会存在重复消费问题。