一文读懂kafka消息丢失问题和解决方案

约定不等于承诺〃 2023-09-25 18:42 51阅读 0赞

## 前言 ##

今天分享一下kafka的消息丢失问题，kafka的消息丢失是一个很值得关注的问题，根据消息的重要性，消息丢失的严重性也会进行放大，如何从最大程度上保证消息不丢失，要从生产者，消费者，broker几个端来说。

## 消息发送和接收流程 ##

kafka生产者生产好消息后，会将消息发送到broker节点，broker对数据进行存储，kafka的消息是顺序存储在磁盘上，以主题(topic)，分区(partition)的逻辑进行划分，消息最终存储在日志文件中，消费者会循环从broker拉取消息。

![format_png][]

那么从上图的图中可以看出kafka丢消息可能存在的三个地方分别为：

*  生产者到broker
 *  broker到磁盘
 *  消费者

## 生产者到broker消息丢失 ##

生产者发送消息到broker是会存在消息丢失的，大多可能是由于网络原因引起的，消息中间件中一般都是通过ack来解决这个问题的，kafka中可以通过设置ack来解决这个问题。

acks有三种类型：

*  0
 *  1
 *  \-1(all)

### acks为0 ###

acks设置为0，代表生产者发送消息后就不管不顾了，不用等待broker的任何响应，那么可能网络异常或者其他原因导致broker没有处理到到这条消息，那么消息就丢失了。

![format_png 1][]

### acks为1 ###

acks设置为1，代表生产者发送消息到broker后，只需要broker的leader副本确认收到后就成功响应，不需要follower副本响应，就算follower副本崩溃了，也会成功响应。

![format_png 2][]

### acks为-1(all) ###

acks设置为-1，或者为all，那么生产者发送消息需要leader和follower都收到并写入消息才成功响应生产者，也就是`ISR`集合要全部写入，当ISR集合中只要有一个没有写入成功，那么就收到失败响应，所以acks=-1能够在最大程度上保证消息的不丢失，`但是也是有条件的，需要ISR集合中有两个以上副本才能保证，如果只有一个副本，那么就是就只有一个leader，没有follower，如果leader挂掉，就不能选举出一个eader，消息自然也就丢失，这和acks=1是一样的`。

![format_png 3][]

## 解决消息丢失 ##

从上面三种类型的acks中我们可以看出，acks=-1是保证消息从生产者到broker不丢失的最佳设置方式，不过我们也能想到，它需要ISR每个副本都成功应答，所以它的效率自然没有前面两个高，不过此篇我们讨论的是保证消息不丢失问题，所以一切从不丢失层面区说。

如果消息发送失败，那么生产者可以重试发送消息，可以手动在代码中编写消息重发逻辑，也可以配置重试参数。

*  retries
 *  retry.backoff.ms

retries表示重试次数，retry.backoff.ms表示重试时间间隔，比如第一次重试依旧没成功，那么隔多久再进行重试，kafka重试的底层逻辑是将没发送成功的消息重新入队，因为kafka的生产者生产消息后，消息并非就直接发送到broker，而是保存在生产者端的`收集器(RecordAccumulator)`，然后由Sender线程去获取RecordAccumulator中的消息，然后再发送给broker，当消息发送失败后，会将消息重新放入RecordAccumulator中，具体逻辑可以看kafka的生产者端Sender的源码。

![format_png 4][]

### 消息重发引起的消息顺序性问题 ###

要注意，消息发送失败进行重发不能保证消息发送的顺序性，`这里的顺序性是单分区顺序性`，如果服务对于消息的顺序性有严格的要求，那么我们可以通过设置属性`max.in.flight.requests.per.connection=1`来保证消息的顺序性，这个配置对应的是kafka中`InFlightRequests`，`max.in.flight.requests.per.connection`代表请求的个数，kafka在创建Sender的时候会判断，如果maxInflightRequests为1，那么guaranteeMessageOrder就为true,就能保证消息的顺序性。

![format_png 5][]

## 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的消息。

![format_png 6][]

那么下次当消费者读取消息的时候是从哪里读取呢，当然从`4`开始读取，因为是从上次读取的offset的下一位开始读取，所以我们就说`当前消费组的offset为4`，，因为下次是从4开始消费，如果5秒之内又消费了两条消息然后自动提交了offset，那么此时的offset如下：

![format_png 7][]

enable.auto.commit如果为false，就代表不会自动提交offset。

auto.offset.reset=latest代表从分区中`最新的offset`处开始读取消息，比如某个消费者组上次提交的偏移量为5，然后后面又生产了2条消息，再次读取消息时，读取到的是`6,7,8`这个三个消息，如果`enable.auto.commit`设置为false，那么不管往分区中写入多少消息，都是从`6`开始读取消息。

![format_png 8][]

此时如果一个`新的的消费组`订阅了这个分区，因为这个消费者组没有在这个分区提交过offset，所以它获取消息并不是从`6`开始获取，而是从`1`开始获取。

![format_png 9][]

> 所以可知每个消费者组在分区中的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到磁盘以及消费者端进行说明，也引申出一些知识点，可能平时没有遇到消息丢失的情况，那是因为网络比较可靠，数据量可能不大，但是如果要真的实现高可用，高可靠，那么就需要对其进行设计。

[format_png]: https://img-blog.csdnimg.cn/img_convert/476c3987a9247b87e9918f72ba9c6cf5.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/c3d8c86a29f6aea66979ba551314d656.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/207699c4a06ce4f40ee23200f8efbd60.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/3aa5b18978ce8e2896df4485ec6080e8.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/c045cac5cab82ce21549dedde552edca.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/f2e1dd986ac33ec61e3647d2920af0bf.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/52ec9748b0ad50cd88c7ea7b91d4556c.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/53baeb6d7f7e7d7ab42f52d3a02718c3.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/28a2a8f4b625f0998cc09013ada6e203.webp?x-oss-process=image/format,png
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/cc1c78d52d94a15c29961f0c91f60b11.webp?x-oss-process=image/format,png