【Kafka】原理分析：详解副本协同流程

红太狼 2022-12-21 06:02 288阅读 0赞

上一篇[【Kafka】原理分析：分区副本机制][Kafka]我们介绍了什么是分区副本机制，里面提到了副本的协同，那么副本间数据同步的具体过程是什么样呢？本篇我们就一起来看看…

## 1.Leader副本接收消息 ##

### 1.1 处理流程 ###

Producer在发布消息到某个Partition时，

1.  寻找Leader：先通过ZooKeeper找到该Partition的Leader `get /brokers/topics/<topic>/partitions/2/state`，然后无论该Topic的Replication Factor为多少（也即该Partition有多少个Replica），Producer**只**将该消息发送到该Partition的Leader。
2.  Leader写入：Leader会将该消息写入其本地**Log**
3.  Follower poll：每个Follower都从Leader pull数据。这种方式上，**Follower 存储的数据顺序与Leader保持一致**。
4.  Followe写入：Follower在收到该消息并写入其Log后，向Leader发送**ACK**（同步也有ack机制）
5.  Leader返回ACK：一旦Leader收到了**ISR中所有**Replica的ACK，该消息就被认为已经commit了，Leader将增加 HW(HighWatermark)并且向Producer发送ACK。

### 1.2 Producer的ack ###

acks参数配置表示producer发送消息到broker上以后的确认机制。

public static final String ACKS_CONFIG = "acks";  // 通过ProducerConfig.ACKS_CONFIG进行配置

*  0：表示producer不需要等待broker的消息确认。这个选项时延最小但同时风险最大（因为当server宕机时，数据将会丢失）
 *  1：表示producer只需要获得kafka集群中的leader节点确认即可，这个选择时延较小同时确保了 leader节点确认接收成功。
 *  all(-1)：需要ISR中所有的Replica给予接收确认，速度最慢，安全性最高，但是由于ISR可能会缩小到仅包含一个Replica，所以设置参数为all并不能一定避免数据丢失，

## 2.Follower副本同步数据 ##

初始状态下，leader和follower的HW和LEO都是0，leader副本会保存remote LEO(远程副本的LEO），表示所有follower LEO，也会被初始化为0，这个时候，producer没有发送消息。follower会不断地给leader发送**FETCH** 请求，但是因为没有数据，这个请求会被leader**寄存**，而寄存时会出现两种情况：

1.  若在指定的时间内没有Producer发送消息，那么leader会强制完成请求。这个时间配置是(replica.fetch.wait.max.ms)
2.  如果在指定时间内producer有消息发送过来，那么kafka会唤醒fetch请求，让leader继续处理

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

对于Producer发送消息后的数据同步，也会出现两种情况：

*  leader处理完producer请求之后，follower发送一个fetch请求过来
 *  follower阻塞在leader指定时间之内，leader副本收到producer的请求（同上2）

这两种情况有着不同的处理方式，下面我们一起来详细的看看。

### 2.1 第一种情况 ###

leader处理完producer请求之后，follower发送一个fetch请求过来。

1). 生产者发送一条消息leader副本：

1.  把消息追加到log文件，同时更新leader副本的LEO
2.  尝试更新leader HW值。这个时候由于follower副本还没有发送fetch请求，那么leader的remote LEO仍然是0。leader会比较自己的LEO以及remote LEO的值发现最小值是0，与HW的值相同，所以不会更新HW

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]

2). 第一次fetch，同步log

1.  leader副本收到Follower发送的fetch请求：
    
    1.  读取log数据，
        
         *  尝试更新remote LEO，因为这个值是根据follower的fetch请求中的oﬀset来确定的，而follower还没有写入这条消息，所以还是0
        
        1.  尝试更新HW，根据remote LEO和LEO取小的，所以仍然是HW=0
    2.  把消息内容和当前分区的HW值发送给follower副本
2.  follower副本收到response：
    
    1.  将消息写入到本地log，同时更新follower的LEO
    2.  尝试更新 follower HW，根据leader HW和本地LEO取小的，所以follower HW 仍然是0

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]

3). 第二次fetch，更新HW

1.  leader收到follower第二次发送的fetch请求后：
    
    1.  读取log数据
        
         *  更新remote LEO=1， 因为这次fetch携带的oﬀset是1.
         *  更新当前分区的HW，这个时候leader LEO和remote LEO都是1，所以HW的值也更新为1
    2.  把数据和当前分区的HW值返回给follower副本，这个时候如果没有数据，则返回为空
2.  follower副本收到response后：
    
    1.  如果有数据则写本地日志，并且更新LEO
    2.  更新follower的HW值

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 3]

到目前为止，数据的同步就完成了，意味着消费端能够消费oﬀset=1这条消息。

### 2.2 第二种情况 ###

前面说过，由于leader副本暂时没有数据过来，所以follower的fetch会被阻塞，直到等待超时或者 leader接收到新的数据当leader收到请求以后会唤醒处于阻塞的fetch请求。处理过程基本上和前面说的一致

1.  leader将消息写入本地日志，更新Leader的LEO
2.  唤醒follower的fetch请求
3.  更新HW

## 3.同步时的数据丢失问题 ##

kafka使用HW和LEO的方式来实现副本数据的同步，本身是一个好的设计，但是在这个地方会存在一个数据丢失的问题，当然这个丢失只出现在特定的背景下。回想一下，HW的值是在新的一轮FETCH 中才会被更新。我们分析下这个过程为什么会出现数据丢失。

### 3.1 数据丢失情况前提 ###

只有同时满足以下两种情况，才有可能出现数据丢失：

*  min.insync.replicas=1 ，设定**ISR中的最小副本**数是多少，默认值为1（在server.properties中配置）
 *  acks参数设置为-1，表示需要所有副本确认（注：必须有ack设置，最小副本数配置才生效）

表达的含义是，至少需要多少个副本同步才能表示消息是提交的， 所以，当 min.insync.replicas=1 的时候，一旦消息被写入leader端log即被认为是“已提交”，而延迟一轮FETCH RPC更新HW值的设计使 得follower HW值是异步延迟更新的，倘若在这个过程中leader发生变更，那么成为新leader的 follower的HW值就有可能是过期的，使得clients端认为是成功提交的消息被删除。

### 3.2 数据丢失流程 ###

1). leader副本本来的LEO=1，现在Producer又发来了一条消息

1.  此时leader副本LEO=2
2.  第一次fetch，follower副本更新LEO=2
3.  第二次fetch，leader更新HW=2，remote LEO=2，

而此时，follower副本在收到响应 response（leader HW=2）前宕机了…

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 4]

2). follower副本恢复，为了数据一致性，首先根据原有的HW=1进行了数据截断，删除了offset=2的消息。这时，在发送fetch请求前leader宕机了…

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 5]

3). leader副本恢复，但此时follower副本已经成为leader，

1.  leader副本因为HW=2，所以不用根据HW截断
2.  leader副本向现leader（原follower）发送fetch，收到Response的HW=1，所以leader副本要截断offset=2的消息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 6]

===> 最终offset=2的消息丢失，其实问题关键就在于follower副本恢复之后进行了截断

### 3.3 数据丢失解决方案 ###

在kafka0.11.0.0版本之后，引入了一个leader epoch来解决这个问题，实际上是一对值`（epoch，oﬀset）`

*  epoch代表leader的版本号，从0开始递增，当leader发生过变更，epoch 就+1
 *  oﬀset则是对应这个epoch版本的leader写入第一条消息的oﬀset，即startOffset

比如 (0,0), (1,50) ，表示第一个leader从oﬀset=0开始写消息，第二个leader版本号是1，从 oﬀset=50开始写；从而可以推出第一个leader写了50条消息。

这个信息会持久化在对应的分区的本地磁盘上，文件名是 `/tmp/kafkalog/topic/leader-epoch-checkpoint` 。leader broker中会保存这样一个缓存，并且定期写入到checkpoint文件中，当leader写log时它会尝试更新整个缓存:

*  如果这个leader首次写消息，则会在缓存中增加一个条目；
 *  否则就不做更新。而每次副本重新成为leader时会查询这部分缓存，获取出对应leader版本的oﬀset

基于同样的情况来分析，follower宕机并且恢复之后，有两种情况：

1.  如果这个时候leader副本没有挂，也就是意味着没有发生leader选举，那么follower恢复之后并不会去截断自己的日志，而是先发送一个OﬀsetsForLeaderEpochRequest请求给到leader副本，leader副本收到请求之后返回当前的 LEO。
    
     *  如果follower副本的leaderEpoch和leader副本的epoch相同， leader的LEO只可能大于或者等于 follower副本的LEO值，所以这个时候不会发生截断
     *  如果follower副本和leader副本的epoch值不同，那么leader副本会查找follower副本传过来的 epoch+1在本地文件中存储的StartOﬀset返回给follower副本，也就是新leader副本的LEO。这样也避免了数据丢失的问题
2.  如果leader副本宕机了重新选举新的leader，那么原本的follower副本就会变成leader，意味着epoch 从0变成1，使得原本follower副本中LEO的值的到了保留。

## 4.副本leader选举 ##

Kafka提供了数据复制算法保证，如果leader副本所在的broker节点宕机或者出现故障，或者分区的 leader节点发生故障，这个时候怎么处理呢？ kafka必须要保证从follower副本中选择一个新的leader副本。

KafkaController会监听ZooKeeper的/brokers/ids节点路径，一旦发现有broker挂了，执行下面的逻辑。这里暂时先不考虑KafkaController所在broker挂了的情况，KafkaController挂了，各个 broker会重新leader选举出新的KafkaController

若leader副本挂了，leader副本在该broker上的分区就要重新进行leader选举，目前的选举策略是：

1.  优先从isr列表中选出第一个作为leader副本，这个叫优先副本，理想情况下有限副本就是该分区的leader副本
2.  如果isr列表为空，则查看该topic的unclean.leader.election.enable配置。
    
     *  unclean.leader.election.enable为true，代表允许选用非isr列表的副本作为leader，那么就意味着数据可能丢失
        
        注：此时会从其他副本中选出一个作为leader副本，并且新的isr列表只包含该leader 副本
     *  unclean.leader.election.enable为false，表示不允许，直接抛出NoReplicaOnlineException异常，造成leader副本选举失败。

一旦选举成功，则将选举后的leader和isr和其他副本信息写入到该分区的对应的zk路径上。

[Kafka]: https://blog.csdn.net/weixin_43935927/article/details/109699182
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/6bd2de3b07284c49971ea36d3f8f4cf5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/5faaf6d5cba944188f7b7cb6eb2c2ab2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/82474bc37ff841bea56ae5e767f5b671.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/908162203bb646c684424bbc85a02fa5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221120/87dfc28233c64eefa6b686ff56b0d43d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221120/37ed458be7104ccd99789be1c0b8f201.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221120/1f48fc0173754fbba193d164713f0107.png