面试官问:Kafka 会不会丢消息?怎么处理的?

沙海
沙海
沙海
718
文章
2
评论
2021年3月9日12:58:06
评论
4 7512字阅读25分2秒
摘要

速读摘要

速读摘要

这个Kafka确实存在丢消息的问题,消息丢失会发生在Broker,Producer和Consumer三种。cache)中,按照时间或者其他条件进行刷盘(从page cache到file),或者通过fsync命令强制刷盘。如果在follower收到数据以后,成功返回ack,leader断电,数据将存在于原来的follower中。在正常情况下,客户端的异步调用可以通过callback来处理消息发送失败或者超时的情况,一旦producer被非法的停止了,那么buffer中的数据将丢失,broker将无法收到该部分数据。

原文约 4549 | 图片 5 | 建议阅读 10 分钟 | 评价反馈

面试官问:Kafka 会不会丢消息?怎么处理的?

搜云库编辑部 搜云库技术团队

周一,中午好,我是磊哥。

今天,磊哥给大家科普一下,面试官经常问的刁钻问题,Kafka 会不会丢消息?怎么处理的?

这个Kafka确实存在丢消息的问题,消息丢失会发生在Broker,Producer和Consumer三种。

面试官问:Kafka 会不会丢消息?怎么处理的?

Broker

Broker丢失消息是由于Kafka本身的原因造成的,kafka为了得到更高的性能和吞吐量,将数据异步批量的存储在磁盘中。消息的刷盘过程,为了提高性能,减少刷盘次数,kafka采用了批量刷盘的做法。即,按照一定的消息量,和时间间隔进行刷盘。这种机制也是由于linux操作系统决定的。将数据存储到linux操作系统种,会先存储到页缓存(Page cache)中,按照时间或者其他条件进行刷盘(从page cache到file),或者通过fsync命令强制刷盘。数据在page cache中时,如果系统挂掉,数据会丢失。

面试官问:Kafka 会不会丢消息?怎么处理的?

Broker在linux服务器上高速读写以及同步到Replica

上图简述了broker写数据以及同步的一个过程。broker写数据只写到PageCache中,而pageCache位于内存。这部分数据在断电后是会丢失的。pageCache的数据通过linux的flusher程序进行刷盘。刷盘触发条件有三:

Broker配置刷盘机制,是通过调用fsync函数接管了刷盘动作。从单个Broker来看,pageCache的数据会丢失。

Kafka没有提供同步刷盘的方式。同步刷盘在RocketMQ中有实现,实现原理是将异步刷盘的流程进行阻塞,等待响应,类似ajax的callback或者是java的future。下面是一段rocketmq的源码。

GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());service.putRequest(request);boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout()); // 刷盘

也就是说,理论上,要完全让kafka保证单个broker不丢失消息是做不到的,只能通过调整刷盘机制的参数缓解该情况。比如,减少刷盘间隔,减少刷盘数据量大小。时间越短,性能越差,可靠性越好(尽可能可靠)。这是一个选择题。

为了解决该问题,kafka通过producer和broker协同处理单个broker丢失参数的情况。一旦producer发现broker消息丢失,即可自动进行retry。除非retry次数超过阀值(可配置),消息才会丢失。此时需要生产者客户端手动处理该情况。那么producer是如何检测到数据丢失的呢?是通过ack机制,类似于http的三次握手的方式。

The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent. The following settings are allowed: acks=0 If set to zero then the producer will not wait for any acknowledgment from the server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case, and the retries configuration will not take effect (as the client won’t generally know of any failures). The offset given back for each record will always be set to -1. acks=1 This will mean the leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost. acks=allThis means the leader will wait for the full set of in-sync replicas to acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee. This is equivalent to the acks=-1 setting.

以上的引用是kafka官方对于参数acks的解释(在老版本中,该参数是request.required.acks)。

上面第三点提到了ISR的列表的follower,需要配合另一个参数才能更好的保证ack的有效性。ISR是Broker维护的一个“可靠的follower列表”,in-sync Replica列表,broker的配置包含一个参数:min.insync.replicas。该参数表示ISR中最少的副本数。如果不设置该值,ISR中的follower列表可能为空。此时相当于acks=1。

面试官问:Kafka 会不会丢消息?怎么处理的?

如上图中:

  • acks=0,总耗时f(t) = f(1)。

  • acks=1,总耗时f(t) = f(1) + f(2)。

  • acks=-1,总耗时f(t) = f(1) + max( f(A) , f(B) ) + f(2)。

性能依次递减,可靠性依次升高。

Producer

Producer丢失消息,发生在生产者客户端。

为了提升效率,减少IO,producer在发送数据时可以将多个请求进行合并后发送。被合并的请求咋发送一线缓存在本地buffer中。缓存的方式和前文提到的刷盘类似,producer可以将请求打包成“块”或者按照时间间隔,将buffer中的数据发出。通过buffer我们可以将生产者改造为异步的方式,而这可以提升我们的发送效率。

但是,buffer中的数据就是危险的。在正常情况下,客户端的异步调用可以通过callback来处理消息发送失败或者超时的情况,但是,一旦producer被非法的停止了,那么buffer中的数据将丢失,broker将无法收到该部分数据。又或者,当Producer客户端内存不够时,如果采取的策略是丢弃消息(另一种策略是block阻塞),消息也会被丢失。抑或,消息产生(异步产生)过快,导致挂起线程过多,内存不足,导致程序崩溃,消息丢失。

面试官问:Kafka 会不会丢消息?怎么处理的?

producer采取批量发送的示意图

面试官问:Kafka 会不会丢消息?怎么处理的?

异步发送消息生产速度过快的示意图

根据上图,可以想到几个解决的思路:

Consumer

Consumer消费消息有下面几个步骤:

Consumer的消费方式主要分为两种:

Consumer自动提交的机制是根据一定的时间间隔,将收到的消息进行commit。commit过程和消费消息的过程是异步的。也就是说,可能存在消费过程未成功(比如抛出异常),commit消息已经提交了。此时消息就丢失了。

Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "test");// 自动提交开关props.put("enable.auto.commit", "true");// 自动提交的时间间隔,此处是1sprops.put("auto.commit.interval.ms", "1000");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Arrays.asList("foo", "bar"));while (true) {        // 调用poll后,1000ms后,消息状态会被改为 committed  ConsumerRecords<String, String> records = consumer.poll(100);  for (ConsumerRecord<String, String> record : records)    insertIntoDB(record); // 将消息入库,时间可能会超过1000ms

上面的示例是自动提交的例子。如果此时,insertIntoDB(record)发生异常,消息将会出现丢失。接下来是手动提交的例子:

Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "test");// 关闭自动提交,改为手动提交props.put("enable.auto.commit", "false");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Arrays.asList("foo", "bar"));final int minBatchSize = 200;List<ConsumerRecord<String, String>> buffer = new ArrayList<>();while (true) {        // 调用poll后,不会进行auto commit  ConsumerRecords<String, String> records = consumer.poll(100);  for (ConsumerRecord<String, String> record : records) {    buffer.add(record);  }  if (buffer.size() >= minBatchSize) {    insertIntoDb(buffer);                // 所有消息消费完毕以后,才进行commit操作    consumer.commitSync();    buffer.clear();  }

将提交类型改为手动以后,可以保证消息“至少被消费一次”(at least once)。但此时可能出现重复消费的情况,重复消费不属于本篇讨论范围。

上面两个例子,是直接使用Consumer的High level API,客户端对于offset等控制是透明的。也可以采用Low level API的方式,手动控制offset,也可以保证消息不丢,不过会更加复杂。

 try {     while(running) {         ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);         for (TopicPartition partition : records.partitions()) {             List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);             for (ConsumerRecord<String, String> record : partitionRecords) {                 System.out.println(record.offset() + ": " + record.value());             }             long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();             // 精确控制offset             consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));         }     } } finally {   consumer.close(); }

来源 | https://blog.dogchao.cn/?p=305

近期技术热文1、面试官欺负人:new Object()占用几个字节?2、不要封装工具类了,这款神仙框架,真好用!3、退税 = 磊哥发财了?4、磊哥,做副业=9034.75元?5、Java 8 中 Map 骚操作之 merge() 的用法推荐,发现一个 GitHub 书籍仓库https://github.com/itdevbooks/pdf这个整理了1000多本 常用 技术书籍PDF,绝大部分核心的高清技术书籍都可以在这里找到!推荐,GitHub 地址,电脑打开体验更好阅读原文:一键直达,GitHub 地址

阅读原文

继续阅读
weinxin
资源分享QQ群
本站是一个IT技术分享社区, 会经常分享资源和教程; 分享的时代, 请别再沉默!
沙海
匿名

发表评论

匿名网友 填写信息

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen: