浅谈Kafka中acks参数对消息持久化的影响

た入场券 2022-11-29 11:30 123阅读 0赞

在说到Kafka的参数优化时就给大家说过acks参数的意义，这个acks参数在kafka的使用中，是非常核心以及关键的一个参数，决定了很多东西。尤其时数据丢失这方面，能很好解决这类问题。  
为了保证producer发送的数据能可靠的发送到指定的topic，topic的每个partition收到producer发送的数据后都需要向producer发送确认消息，即acknowledgement确认收到消息，如果producer收到ack就会进行下一轮的发送，否则重新发送数据。  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center][]  
那么所在的broker什么时候发送ack给producer？

1.  leader落盘后直接返回ack，如果producer收到ack后leader挂了，follower还未同步成功，造成消息丢失。
2.  follower全部同步成功后返回ack，leader选举时容忍n台节点故障，需要n+1个副本，随便选一个follower即可。不过是高延迟，需要等待follower都同步成功。
3.  半数以上follower同步成功后返回ack，选举新leader时，容忍n台节点故障，需要2n+1个副本。

kafka选择所有follower同步成功后再返回ack的方案，原因如下：

*  同样为了容忍n台节点故障，第一种方案需要2n+1个副本，而第二种方案只需要n+1个副本，而kafka每个分区都有大量的数据，第一种方案会造成大量的数据冗余。
 *  第二种方案的网络延迟虽然比较高，但都是一个局域网中，网络延时的影响也比较小。

**多副本冗余机制**  
比如Partition0有一个副本是Leader，另外一个副本是Follower，Leader和Follower两个副本是分布在不同机器上的。这样的多副本冗余机制，可以保证任何一台机器挂掉，都不会导致数据彻底丢失，因为起码还是有副本在别的机器上的。

**多副本数据同步**  
接着我们就来看看多个副本之间数据是如何同步的？其实任何一个Partition，只有Leader是对外提供读写服务的。  
也就是说，如果有一个客户端往一个Partition写入数据，此时一般就是写入这个Partition的Leader副本。  
然后Leader副本接收到数据之后，Follower副本会不停的给他发送请求尝试去拉取最新的数据，拉取到自己本地后，写入磁盘中。

**ISR含义和作用**  
ISR(In-Sync Replica)：是Replicas的一个子集，表示目前Alive且与Leader能够“Catch-up”的Replicas集合。也就是保持同步的副本，他的含义就是，跟Leader始终保持同步的Follower有哪些。

读写都是首先落到Leader上，所以一般来说通过同步机制从Leader上拉取数据的Replica都会和Leader有一些延迟(包括了延迟时间和延迟条数两个维度)，任意一个超过阈值都会把该Replica踢出ISR。每个Partition都有它自己独立的ISR。  
大家可以想一下 ，如果说某个Follower所在的Broker因为JVM FullGC之类的问题，导致自己卡顿了，无法及时从Leader拉取同步数据，那么是不是会导致Follower的数据比Leader要落后很多？

所以这个时候，就意味着Follower已经跟Leader不再处于同步的关系了。但是只要Follower一直及时从Leader同步数据，就可以保证他们是处于同步的关系的。  
所以每个Partition都有一个ISR，这个ISR里一定会有Leader自己，因为Leader肯定数据是最新的，然后就是那些跟Leader保持同步的Follower，也会在ISR里。

leader维护了一个动态的in-sync-replica (与leader保持同步的副本集合，该集合中的副本和leader数据是一致的)，当ISR中的follower和leader数据同步完成之后，leader就会向follower发送ack，如果ISR中的follower长时间未向leader发送同步完成消息，leader会将其从ISR中剔除，等待的时长由replica.time.max.ms参数设定。leader故障之后，就会从ISR中选举出新的leader。

**acks参数的含义**  
如果大家没看明白前面的那些副本机制、同步机制、ISR机制，那么就无法充分的理解acks参数的含义，这个参数实际上决定了很多重要的东西。

首先这个acks参数，是在KafkaProducer，也就是生产者客户端里设置的；  
也就是说，你往kafka写数据的时候，就可以来设置这个acks参数。然后这个参数实际上有三种常见的值可以设置，分别是：0、1 、-1和 all。

**第一种选择是把acks参数设置为0**  
**acks=0**  
意思就是我的KafkaProducer在客户端，只要把消息发送出去，不管发送出去的数据有没有同步完成，都不用管，直接就认为这个消息发送成功了。意味着producer不等待broker同步完成的确认，继续发送下一条(批)信息，提供了最低的延迟，但是持久性最差；当服务器发生故障时，就很可能发生数据丢失。例如leader已经死亡，producer不知情，还会继续发送消息broker接收不到数据就会数据丢失。  
就比如可能你发送出去的消息还在半路，结果呢，Partition Leader所在的Broker就直接挂了，然后你的客户端还认为消息发送成功了，此时就会导致这条消息就丢失了。

**第二种选择是把acks参数设置为1（默认）**  
**acks=1**  
意思就是说只要Partition Leader接收到消息而且写入本地磁盘了，就认为成功了，不管其他的Follower有没有同步过去这条消息了。**这种设置其实是kafka默认的设置**，大家请注意，划重点！这是默认的设置，也就是说，默认情况下，你要是不重新设置acks这个参数，只要Partition Leader写成功导磁盘就算成功。  
但是这里有一个问题，万一Partition Leader刚刚接收到消息，Follower还没来得及同步过去，结果Leader所在的broker宕机了，此时也会导致这条消息丢失，因为客户端已经认为发送成功了。  
意味着producer要等待leader成功收到数据并得到确认，才发送下一条message。此选项提供了较好的持久性较低的延迟性。但如果Partition的Leader死亡，follwer尚未复制，数据就会丢失。

**第三种选择是把acks参数设置为all**  
**acks=all**  
这个意思就是说，Partition Leader接收到消息之后，还必须要求ISR列表里跟Leader保持同步的那些Follower都要把消息同步过去，才能认为这条消息是写入成功了。

如果说Partition Leader刚接收到了消息，但是结果Follower没有收到消息，此时Leader宕机了，那么客户端会感知到这个消息没发送成功，他会重试再次发送消息过去。

此时可能Partition2的Follower变成Leader了，此时ISR列表里只有最新的这个Follower转变成的Leader了，那么只要这个新的Leader接收消息就算成功了。意味着producer得到follwer确认，才发送下一条数据。持久性最好，延时性最差。

**参数设置成-1(或者是all)**  
producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。  
但是这样也不能保证数据不丢失，比如当ISR中只有leader时（前面ISR讲到，ISR中的成员由于某些情况会增加也会减少，最少就只剩一个leader），这样就变成了acks=1的情况。

如果要提高数据的可靠性，在设置request.required.acks=-1的同时，也要min.insync.replicas这个参数(可以在broker或者topic层面进行设置)的配合，这样才能发挥最大的功效。

min.insync.replicas这个参数设定ISR中的最小副本数是多少，默认值为1，当且仅当request.required.acks参数设置为-1时，此参数才生效。如果ISR中的副本数少于min.insync.replicas配置的数量时，客户端会返回异常：

org.apache.kafka.common.errors.NotEnoughReplicasExceptoin: Messages are rejected since there are fewer in-sync replicas than required

**接下来对acks=1和-1的两种情况进行详细分析：**

*  **request.required.acks=1**  
    producer发送数据到leader，leader写本地日志成功，返回客户端成功；此时ISR中的副本还没有来得及拉取该消息，leader就宕机了，那么此次发送的消息就会丢失。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]
 *  **request.required.acks=-1**  
    同步（Kafka默认为同步，即producer.type=sync）的发送模式，replication.factor>=2且min.insync.replicas>=2的情况下，不会丢失数据。  
    有两种典型情况。**acks=-1**的情况下（如无特殊说明，以下acks都表示为参数request.required.acks），数据发送到leader, ISR的follower全部完成数据同步后，leader此时挂掉，那么会选举出新的leader，数据不会丢失。如下图所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 2]  
**acks=-1**的情况下，数据发送到leader后 ，部分ISR的副本同步，leader此时挂掉。比如follower1和follower2都有可能变成新的leader, producer端会得到返回异常，producer端会重新发送数据，数据可能会重复。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 3]  
当然上图中如果在leader crash的时候，follower2还没有同步到任何数据，而且follower2被选举为新的leader的话，这样消息就不会重复。

**acks=all 就可以代表数据一定不会丢失了吗？**  
当然不是，如果你的Partition只有一个副本，也就是一个Leader，任何Follower都没有，你认为acks=all有用吗？  
当然没用了，因为ISR里就一个Leader，他接收完消息后宕机，也会导致数据丢失。  
所以说，这个acks=all，必须跟ISR列表里至少有2个以上的副本配合使用，起码是有一个Leader和一个Follower才可以。  
这样才能保证说写一条数据过去，一定是2个以上的副本都收到了才算是成功，此时任何一个副本宕机，不会导致数据丢失。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/04dd6e7e5597404aad9ac92a99691280.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/682f988e1b424309910f01ce5016f158.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221124/6e25a9c1bec24e95be6f07a33be7c1fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221124/8a7b7c2bff114cf9990d493f99d27613.png