Kafka~消息系列问题解决：消费顺序问题解决、消息丢失问题优化（不能保证100%）

消息消费顺序问题

使用消息队列的过程中经常有业务场景需要严格保证消息的消费顺序，比如我们同时发了 2 个消息，这 2 个消息对应的操作分别对应的数据库操作是：

1. 用户等级升级。
2. 根据用户等级下的订单价格

假如这两条消息的消费顺序不一样造成的最终结果就会截然不同。我们知道 Kafka 中 Partition(分区)是真正保存消息的地方，我们发送的消息都被放在了这里。而我们的 Partition(分区) 又存在于 Topic(主题) 这个概念中，并且我们可以给特定 Topic 指定多个 Partition。每次添加消息到 Partition(分区) 的时候都会采用尾加法。

Kafka 只能为我们保证 Partition(分区) 中的消息有序。消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。

为什么同一个partition的消息是有序的?

• 因为当生产者向某个partition发送消息时，消息会被追加到该ppartition的日志文件(log)中，并且被分配一个唯一的offset，文件的读写是有顺序的。而消费者在从该分区消费消息时，会从该分区的最早offset开始逐个读取 消息，保证了消息的顺序性。

如何解决

Kafka 通过偏移量（offset）来保证消息在分区内的顺序性。所以，我们就有一种很简单的保证消息消费顺序的方法： 1 个 Topic 只对应一个 Partition。这样当然可以解决问题，但是破坏了 Kafka 的设计初衷。

Kafka 中发送 1 条消息的时候，可以指定 topic, partition, key,data（数据） 4 个参数。如果你发送消息的时候指定了 Partition 的话，所有消息都会被发送到指定的 Partition。并且，同一个 key 的消息可以保证只发送到同一个 partition，这个我们可以采用表/对象的 id 来作为 key 。

总结一下，对于如何保证 Kafka 中消息消费的顺序，有了下面两种方法：

• 1 个 Topic 只对应一个 Partition。
• （推荐）发送消息的时候指定 key/Partition。

如何发到同一个partition？

当我们发送消息的时候，如果key为null，那么Kafka默认采用Roiund-robin策略，也就是轮转，实现类是DefaultPartitioner。那么如果想要指定他发送到某个partition的话，有以下三个方式:

1. 指定partitionID：我们可以在发送消息的时候，可以直接在ProducerRecord中指定partition
2. 指定key：在没有指定Partition(null值)时，如果有 Key，Kafka将依据Key做hash来计算出一个Partition编号来。如果key相同，那么也能分到同一个partition中，但这个方案如果后续要增加删除partition就会出现短暂的乱序。
3. 自定义Partitioner：可以实现自己的分区器(Partitioner)来指定消息发送到特定的分区。我们需要创建一个类实现Partitioner接口，并且重写partition()方法。

消息丢失问题

生产者丢失问题调优

生产者(Producer) 调用send方法发送消息之后，消息可能因为网络问题并没有发送过去。所以，我们不能默认在调用send方法发送消息之后消息发送成功了。为了确定消息是发送成功，我们要判断消息发送的结果。

但是要注意的是 Kafka 生产者(Producer) 使用 send 方法发送消息实际上是异步的操作，我们可以通过 get()方法获取调用结果，但是这样也让它变为了同步操作，不推荐这么做！可以采用为其添加回调函数的形式。

如果消息发送失败的话，我们检查失败的原因之后重新发送即可！另外，这里推荐为 Producer 的retries（重试次数）设置一个比较合理的值，一般是 3 ，但是为了保证消息不丢失的话一般会设置比较大一点。设置完成之后，当出现网络问题之后能够自动重试消息发送，避免消息丢失。另外，建议还要设置重试间隔，因为间隔太小的话重试的效果就不明显了，网络波动一次你 3 次一下子就重试完了。

同时,我们也可以通过给producer设置一些参数来提升发送成功率:

• retries=3 //生产端的重试次数
• retry.backoff.ms = 300 //消息发送超时或失败后,间隔的重试时间
• acks=0:表示Producer请求立即返回，不需要等待Leader的任何可确认。这种方案有最高的吞吐率，但是不保证消息是否真的发送成功。
• acks=-1:表示分区Leader必须等待消息被成功写入到所有的ISR副本(同步副本)中才认为Producer请求成功。这种方案提供最高的消息持久性保证，但是理论上吞吐率也是最差的。
• acks=1:表示Leader必须应答此Producer请求并写入消息到本地日志，之后Producer请求被认为成功。如果此时Leader副本应答请求之后挂掉了，消息会丢失。这个方案，提供了不错的持久性保证和吞吐。

消费者丢失问题调优

我们知道消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。偏移量（offset)表示 Consumer 当前消费到的 Partition(分区)的所在的位置。Kafka 通过偏移量（offset）可以保证消息在分区内的顺序性。

当消费者拉取到了分区的某个消息之后，消费者会自动提交了 offset。自动提交的话会有一个问题，试想一下，当消费者刚拿到这个消息准备进行真正消费的时候，突然挂掉了，消息实际上并没有被消费，但是 offset 却被自动提交了。

解决办法也比较粗暴，我们手动关闭自动提交 offset，每次在真正消费完消息之后再自己手动提交 offset 。 但是，细心的朋友一定会发现，这样会带来消息被重新消费的问题。比如你刚刚消费完消息之后，还没提交 offset，结果自己挂掉了，那么这个消息理论上就会被消费两次。

Broker丢消息问题调优

Kafka 为分区（Partition）引入了多副本（Replica）机制。分区（Partition）中的多个副本之间会有一个叫做 leader 的家伙，其他副本称为 follower。我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。

生产者和消费者只与 leader 副本交互。你可以理解为其他副本只是 leader 副本的拷贝，它们的存在只是为了保证消息存储的安全性。

试想一种情况：假如 leader 副本所在的 broker 突然挂掉，那么就要从 follower 副本重新选出一个 leader ，但是 leader 的数据还有一些没有被 follower 副本的同步的话，就会造成消息丢失。

• 设置 acks = all：解决办法就是我们设置 acks = all。acks 是 Kafka 生产者(Producer) 很重要的一个参数。

acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后就算被成功发送。当我们配置 acks = all 表示只有所有 ISR 列表的副本全部收到消息时，生产者才会接收到来自服务器的响应. 这种模式是最高级别的，也是最安全的，可以确保不止一个 Broker 接收到了消息.，该模式的延迟会很高。

• 设置 replication.factor >= 3

为了保证 leader 副本能有 follower 副本能同步消息，我们一般会为 topic 设置 replication.factor >= 3。这样就可以保证每个 分区(partition) 至少有 3 个副本。虽然造成了数据冗余，但是带来了数据的安全性。

• 设置 min.insync.replicas > 1 一般情况下我们还需要设置 min.insync.replicas> 1 ，这样配置代表消息至少要被写入到 2 个副本才算是被成功发送。min.insync.replicas 的默认值为 1 ，在实际生产中应尽量避免默认值 1。但是，为了保证整个 Kafka 服务的高可用性，你需要确保 replication.factor > min.insync.replicas 。为什么呢？设想一下假如两者相等的话，只要是有一个副本挂掉，整个分区就无法正常工作了。这明显违反高可用性！一般推荐设置成 replication.factor = min.insync.replicas + 1。
• 设置 unclean.leader.election.enable = false

Kafka 0.11.0.0 版本开始 unclean.leader.election.enable 参数的默认值由原来的 true 改为
false

我们最开始也说了我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。多个 follower 副本之间的消息同步情况不一样，当我们配置了 unclean.leader.election.enable = false 的话，当 leader 副本发生故障时就不会从 follower 副本中和 leader 同步程度达不到要求的副本中选择出 leader ，这样降低了消息丢失的可能性。

为何无法保证100%

Kafka提供的Producer和Consumer之间的消息传递保证语义有有三种，所谓消息传递语义，其实就是Kafka的消息交付可靠保障，主要有以下三种:

• At most once 消息可能会丢,但绝不会重复传递
• At least once 消息绝不会丢,但可能会重复传递
• Exactly once 每条消息只会被精确地传递一次。既不会多，也不会少

目前，Kafka默认提供的交付可靠性保障是第二种，即At least once，但是,其实依靠Kafka自身,是没有办法100%保证可靠性的。

介绍了很多调优手段，但本质都是调优，无法100%的保证消息一定发生成功、消费成功。

生产者问题

Kafka允许生产者以异步方式发送消息，这意味着生产者在发送送消息后不会等待确认。当然，我们可以注册一个回调等待消息的成功回调。

但是，如果生产者在发送消息之后，Kafka的集群发生故障或崩溃，而消息尚未被完全写入Kafka的日志中，那么这些消息可能会丢失。虽然后续有可能会重试，但是，如果重试也最终失败了呢？如果这个过程中刚好生产者也崩溃了呢？那就可能会导致没有人知道这个消息失败了，就导致不会重试了。

消费者来说比较简单，只要保证在消息成功时，才提交偏移量量就行了，这样就不会导致消息丢失了。

Broker问题

Kafka使用日志来做消息的持久化的，日志文件是存储在磁盘之上的，但是如果Broker在消息尚未完全写入日志之前崩溃，那么这些消息可能会丢失了。

而且，操作系统在写磁盘之前，会先把数据写入Page Cache中，然后再由操作系统中自己决定什么时候同步到磁盘当中，而在这个过程中，如果还没来得及同步到磁盘中，就直接接宕机了，那这个消息也就丢了。

当然，也可以通过配置log.flush.interval.messages=1，来实现类似于同步刷盘的功能，但是又回到了前面说的情况,还没来得及做持久化，就宕机了。

即使Kafka中引入了副本机制来提升消息的可靠性，但是如果发生同步延迟，还没来及的同步，主副本就挂掉了，那么消息就可能会发生丢失。

这几种情况，只从Broker的角度分析，Broker自身是没办法保证消息不丢失的，但是如果配合Producer，再配合 request.required.acks=-1 这种ACK策略，可以确保消息持久化成功之后，才会ACK给Producer，那么，如果我们的Producer在一定时间段内，没有收到ACK，是可以重新发送的。

但是，这种重新发送，就又回到了我们前面介绍生产者的时候的问题，生产者也有可能挂，重新发送次数也是有上限的，最终发送失败，还是会导致消息最终丢失。

所以，只靠Kafka自己，其实是没有办法保证极端情况下的消息100%不丢失的。

但是，我们也可以在做一些机制来保证，比如引入分布式享事务，或者引入本地消息表等，保证在Kafka Broker没有保存消息成功时，可以重新投递消息。这样才行，类似RocketMQ的实现。