---

怎么保证Kafka的消息不丢失

怎么保证Kafka不丢失消息

Kafka保证不丢失消息需要生产者,Broker及消费者之间的协作,任何一个地方没有可靠性的保证,消息就有可能丢失.因此需要在整个链路上进行全盘考虑,使用具体的配置来达到目的

Broker端配置

• 设置合理的复制因子.Broker可以配置参数replication.factor=N(主题级别,也可以配置Broker级别),配置尽可能大的复制因子意味着分区将会有多个副本,如果副本足够多,那么当一个分区不可用的时候,其它分区就可以接替分区的领导,提高可靠性.
• 禁用不完全的领导选举.Broker可以配置参数unclean.leader.election.enable=false(Broker级别)来禁用不完全的领导选举,这意味着没有完全进行同步的副本不可以选举成为分区的领导,这样不会造成消息的丢失.另外,如果不进行禁用,那么一个处于同步的分区领导A下线之后,另外一个非同步的分区B成为了领导,那么当A上线时,将强制删除之前收到的消息(B并没有这些消息),改为对分区B进行复制,这样造成了消息的丢失.
• 设置合理的最小同步副本个数.Broker可以配置参数min.insync.replicas=N(主题级别和Broker级别)来限制最小同步副本的个数.在Kafka中,当消息被写入到所有同步的副本中,才会被认为是已提交的,如果此时的同步副本只有一个,且配置min.insync.replicas=all,那么只要1个同步副本接收到消息,就认为消息时已提交的.为了保证消息写入到多个同步副本中,我们需要设置一个数量,比如min.insync.replicas=N,当生产者写入消息时,如果此时分区没有N个同步的副本,那么此时消息将不会被写入,生产者将接收到NotEnoughReplicasException.
• 设置合理的心跳超时时间或复制间隔时间. 一个副本被认为不处于同步的情况通常有两种,一种情况是它未能在一定时间内向Zookeeper发送心跳,另外一种情况是在一定时间内未能复制分区领导的消息(或者是在一定时间内未能复制最新的消息).通过配置zookeeper.session.timeout.ms=N,我们可以设置一个合理的心跳超时时间,来减少网络波动或者垃圾回收带来的影响,使得副本变成不同步副本.通过配置replica.lag.time.max.ms,我们可以设置一个合理的复制间隔时间,来使得副本不那么容易变成不同步副本.
• 设置合理的持久化消息大小阈值或者持久化时间间隔.Kafka认为消息是提交的并不需要消息被真正写入到磁盘,因为Kafka认为它的机制已经足够可靠,即使消息没有刷到磁盘中.但是为了进一步提高可靠性,可以通过配置flush.messages,设置一个合理的刷新到磁盘的消息大小数,来保证消息及时刷新到磁盘中.也可以通过配置flush.ms,设置一个合理的刷新到磁盘的时间间隔,保证消息及时刷新到磁盘.

生产者端配置

• 将ack设置为一个合理的值.ack的值代表有多少副本接收到消息,生产者才会收到正确的响应,ack等于0意味着消息被传到网络上,生产者就认为是成功.如果ack=1,就代表一个broker接到消息就认为是成功.将ack=all与min.insync.replicas=N参数结合,可以保证有N个副本收到了消息,此时Broker向生产者返回ack.
• 配置合理的重试参数进行重试.对于Broker返回的可重试的错误,生产者应该进行重试来保证消息不丢失.可以使用默认的最大重试次数(一直重试)配合最大的发送超时时间参数delivery.timout.ms,来保证在超时时间内消息不断进行重试.
• 对于不可重试错误,开发者需要修改配置解决.

消费者端配置

• 最好使用auto.offset.reset=earliest.当消费者消费一个分区时,如果没能获取到上一次同一个消费者组消费该分区的偏移量,那么将采用设置的自动偏移重置参数所配置的策略来重新获取消息,使用earliest虽然可能导致消息被重复消费,但是可以最大限度的保证消息不丢失
• 最好使用手动提交,自动提交有可能提交一些程序还没有处理过的偏移量,使用手动提交会更灵活.
• 成功处理完消息之后再提交偏移量.
• 考虑对分区重平衡进行恰当的处理.分区重平衡时,应该要保证分区在收回之前成功提交偏移量
• 在处理消息失败时进行恰当的处理.当我们在处理消息时发生错误,第一种方式可以使用pause方法暂停在轮询中获取数据,然后重试去处理发生错误的消息.第二种方式可以将发生错误的消息写入到另外一个主题中,然后用另外一个消费者组去单独处理.

参考资料

• Kafka权威指南第二版第七章 Reliable Data Delivery