前言
今天分享一下kafka的消息丢失问题,kafka的消息丢失是一个很值得关注的问题,根据消息的重要性,消息丢失的严重性也会进行放大,如何从最大程度上保证消息不丢失,要从生产者,消费者,broker几个端来说。
消息发送和接收流程
kafka生产者生产好消息后,会将消息发送到broker节点,broker对数据进行存储,kafka的消息是顺序存储在磁盘上,以主题(topic),分区(partition)的逻辑进行划分,消息最终存储在日志文件中,消费者会循环从broker拉取消息。
那么从上图的图中可以看出kafka丢消息可能存在的三个地方分别为:
- 生产者到broker
- broker到磁盘
- 消费者
生产者到broker消息丢失
生产者发送消息到broker是会存在消息丢失的,大多可能是由于网络原因引起的,消息中间件中一般都是通过ack来解决这个问题的,kafka中可以通过设置ack来解决这个问题。
acks有三种类型:
- 0
- 1
- -1(all)
acks为0
acks设置为0,代表生产者发送消息后就不管不顾了,不用等待broker的任何响应,那么可能网络异常或者其他原因导致broker没有处理到到这条消息,那么消息就丢失了。
acks为1
acks设置为1,代表生产者发送消息到broker后,只需要broker的leader副本确认收到后就成功响应,不需要follower副本响应,就算follower副本崩溃了,也会成功响应。
acks为-1(all)
acks设置为-1,或者为all,那么生产者发送消息需要leader和follower都收到并写入消息才成功响应生产者,也就是
ISR
集合要全部写入,当ISR集合中只要有一个没有写入成功,那么就收到失败响应,所以acks=-1能够在最大程度上保证消息的不丢失,
但是也是有条件的,需要ISR集合中有两个以上副本才能保证,如果只有一个副本,那么就是就只有一个leader,没有follower,如果leader挂掉,就不能选举出一个eader,消息自然也就丢失,这和acks=1是一样的
。
解决消息丢失
从上面三种类型的acks中我们可以看出,acks=-1是保证消息从生产者到broker不丢失的最佳设置方式,不过我们也能想到,它需要ISR每个副本都成功应答,所以它的效率自然没有前面两个高,不过此篇我们讨论的是保证消息不丢失问题,所以一切从不丢失层面区说。
如果消息发送失败,那么生产者可以重试发送消息,可以手动在代码中编写消息重发逻辑,也可以配置重试参数。
- retries
- retry.backoff.ms
retries表示重试次数,retry.backoff.ms表示重试时间间隔,比如第一次重试依旧没成功,那么隔多久再进行重试,kafka重试的底层逻辑是将没发送成功的消息重新入队,因为kafka的生产者生产消息后,消息并非就直接发送到broker,而是保存在生产者端的
收集器(RecordAccumulator)
,然后由Sender线程去获取RecordAccumulator中的消息,然后再发送给broker,当消息发送失败后,会将消息重新放入RecordAccumulator中,具体逻辑可以看kafka的生产者端Sender的源码。
消息重发引起的消息顺序性问题
要注意,消息发送失败进行重发不能保证消息发送的顺序性,
这里的顺序性是单分区顺序性
,如果服务对于消息的顺序性有严格的要求,那么我们可以通过设置属性
max.in.flight.requests.per.connection=1
来保证消息的顺序性,这个配置对应的是kafka中
InFlightRequests
,
max.in.flight.requests.per.connection
代表请求的个数,kafka在创建Sender的时候会判断,如果maxInflightRequests为1,那么guaranteeMessageOrder就为true,就能保证消息的顺序性。
broker到磁盘丢消息
broker收到消息后,需要将消息写入磁盘的log文件中,但是并不是马上写,因为我们知道,生产者发送消息后,消费者那边需要马上获取,如果broker要写入磁盘,那么消费者拉取消息,broker还要从log文件中获取消息,这显然是不合理的,所以kafka引入了(page cache)页缓存。
page cache是磁盘和broker之间的消息映射关系,它是基于内存的,当broker收到消息后,会将消息写入page cache,然后由操作系统进行刷盘,将page cache中的数据写入磁盘。
如果broker发生故障,那么此时page cache的数据就会丢失,broker端可以设置刷盘的参数,比如多久刷盘一次,不过这个参数不建议去修改,最好的方案还是设置多副本,一个分区设置几个副本,当broker故障的时候,如果还有其他副本,那么数据就不会丢失。
消费者丢消息
kafka的消费模式是拉模式,需要不断地向broker拉取消息,拉取的消息消费了以后需要提交offset,也就是提交offset这里可能会出现丢消息,kafka中提供了和offset相关的几个配置项。
- enable.auto.commit
- auto.commit.interval.ms
- auto.offset.reset
下面我们先了解一下kafka offset的提交和参数详解。
enable.auto.commit代表是否自动提交offset,默认为true,auto.commit.interval.ms代表多久提交一次offset,默认为5秒。
如下图,当前消费者消费到了分区中为3的消息。
那么下次当消费者读取消息的时候是从哪里读取呢,当然从
4
开始读取,因为是从上次读取的offset的下一位开始读取,所以我们就说
当前消费组的offset为4
,,因为下次是从4开始消费,如果5秒之内又消费了两条消息然后自动提交了offset,那么此时的offset如下:
enable.auto.commit如果为false,就代表不会自动提交offset。
auto.offset.reset=latest代表从分区中
最新的offset
处开始读取消息,比如某个消费者组上次提交的偏移量为5,然后后面又生产了2条消息,再次读取消息时,读取到的是
6,7,8
这个三个消息,如果
enable.auto.commit
设置为false,那么不管往分区中写入多少消息,都是从
6
开始读取消息。
此时如果一个
新的的消费组
订阅了这个分区,因为这个消费者组没有在这个分区提交过offset,所以它获取消息并不是从
6
开始获取,而是从
1
开始获取。
所以可知每个消费者组在分区中的offset是独立的。
auto.offset.reset还可以设置为
earliest
和
none
,使用earliest,
如果此消费组从来没有提交过offset,那么就从头开始消费,如果提交过offset,那么就从最新的offset处消费,就和latest一样了
,使用none,如果消费组没有提交过offset,在分区中找不到任何offset,那么就会抛出异常。
org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partitions: [stock1-0]
复制代码
上面我们初步了解了offset的一些知识,对offset的提交和和读取有一些了解,因为上面我们只提及offset的自动提交,而自动提交的主动权在kafka,而不在我们,所以可能因为一些原因而导致消息丢失。
消息处理异常
当我们收到消息后对消息进行处理,如果在处理的过程中发生异常,而又设置为自动提交offset,那么消息没有处理成功,offset已经提交了,当下次获取消息的时候,由于已经提交过ofset,所以之前的消息就获取不到了,所以应该改为手动提交offset,当消息处理成功后,再进行手动提交offset。
总结
关于kafka的消息丢失问题和解决方案就说到这里,我们分别从生产者到broker,broker到磁盘以及消费者端进行说明,也引申出一些知识点,可能平时没有遇到消息丢失的情况,那是因为网络比较可靠,数据量可能不大,但是如果要真的实现高可用,高可靠,那么就需要对其进行设计。
版权归原作者 肥肥技术宅 所有, 如有侵权,请联系我们删除。