面试系列-kafka高可用机制

用户4283147

发布于 2022-12-29 20:02:54

3990

发布于 2022-12-29 20:02:54

消息备份

Kafka允许同⼀个Partition存在多个消息副本(Replica)，每个Partition的副本通常由1个Leader及0个以上的Follower组成，⽣产者将消息直接发往对应Partition的Leader，Follower会周期地向Leader发送同步请求，Kafka的Leader机制在保障数据⼀致性地同时降低了了消息备份的复杂度；同⼀Partition的Replica不应存储在同一个Broker上，因为一旦该Broker宕机，对应Partition的所有Replica都无法⼯作，这就达不到高可用的效果。为了做好负载均衡并提⾼容错能力，Kafka会尽量将所有的Partition以及各Partition的副本均匀地分配到整个集群上；

ISR机制

kafka中每一个主题又进一步划分成若干个分区。副本的概念实际上是在分区层级下定义的，每个分区配置有多若干个副本；所谓副本，本质上就是一个只能追加写消息的提交日志，根据kafka副本机制的定义，同一个分区下的所有副本保存着相同的消息序列，这些副本分散的保存在不同的Broker上，从而能够对抗部分Broker宕机带来的数据不可用；

在kafka分区中的副本机制中，又分了Leader节点和Follower节点，消息会写到Leader节点中，由Leader节点将数据同步给Follower节点，每个副本都可以进行读操作，从而减轻读的压力；

ISR副本集合

public class FetchRequest {
    private final short versionId;
    private final int correlationId;
    private final String clientId;
    private final int replicaId;
    //Follower容忍的最⼤大等待时间: 到点Leader⽴立即返回结果，默认值500
    private final int maxWait; 
    //Follower容忍的最⼩小返回数据⼤大⼩小:当Leader有⾜足够数据时⽴立即返回，
    //兜底等待
    //Follower中各Partititon
    private final Map<TopicAndPartition, PartitionFetchInfo> requestInfo; 
    private final int minBytes; 
}

ISR中的副本都是与Leader同步的副本集合（实际存储的是副本所在Broker的BrokerId），相反不在ISR中的追随者副本就被认为是与Leader不同步的这里的保持同步不是指与Leader数据保持完全一致，只需在replica.lag.time.max.ms时间内与Leader保持有效连接Follower周期性地向Leader发送FetchRequest请求，发送时间间隔配置在replica.fetch.wait.max.ms中，默认值为500ms；各Partition的Leader负责维护ISR列表并将ISR的变更同步至ZooKeeper，被移出ISR的Follower会继续向Leader发FetchRequest请求，试图再次跟上Leader重新进入ISR，ISR中所有副本都跟上了Leader，通常只有ISR里的成员才可能被选为Leader；

ISR选举机制

在kafka中分区副本选举机制采用的ISR的机制，全称为In-Sync Replicas，挑选出Leader节点，而ISR其实就是一个存放分区副本ID的集合，如果某个副本所在的Broker正常的和zookeepeer能够建立连接的情况下，那这个副本的ID就会存放到该集合中，如果某个副本节点宕机之后，该副本数据就会从该ISR集合中剔除；

当Leader收到消息数据同步给Follower节点时，如果Follower节点一直阻塞，长时间不给Leader节点发送ack ，这种情况也会从ISR集合中剔除；

如果Leader节点宕机之后，其他副本就会将ISR中的Leader节点移除，而ISR列表中最前面的副本就被选举为一个新的Leader节点；

unclean领导者选举

当Kafka中unclean.leader.election.enable配置为true(默认值为false)且ISR中所有副本均宕机的情况下，才允许ISR外的副本被选为Leader，此时会丢失部分已应答的数据；

开启Unclean领导者选举可能会造成数据丢失，但好处是它使得分区Leader副本一直存在，不至于停止对外提供服务，因此提升了高可用性；反之禁止Unclean领导者选举的好处在于维护了数据的一致性，避免了消息丢失，但牺牲了高可用性；

Acks

acks = 0

⽣产者⽆需等待服务端的任何确认，消息被添加到生产者套接字缓冲区后就视为已发送，因此acks=0不能保证服务端已收到消息，使用场景较少；

acks = 1

Leader将消息写入本地日志后无需等待Follower的消息确认就做出应答；如果Leader在应答消息后立即宕机且其他Follower均未完成消息的复制，则该条消息将丢失；

acks = all（-1）

Leader将等待ISR中的所有副本确认后再做出应答，因此只要ISR中任何一个副本还存活着，这条应答过的消息就不会丢失，acks=all是可用性最⾼的选择，但等待Follower应答引入了额外的响应时间，Leader需要等待ISR中所有副本做出应答，此时响应时间取决于ISR中最慢的那台机器；

如果说Partition Leader刚接收到了消息，但是结果Follower没有收到消息，此时Leader宕机了，那么客户端会感知到这个消息没发送成功，会重试再次发送消息过去；Broker有个配置项min.insync.replicas(默认值为1)代表了正常写入生产者数据所需要的最少ISR个数当ISR中的副本数量小于min.insync.replicas时，Leader停止写入生产者生产的消息，并向生产者抛出NotEnoughReplicas异常，阻塞等待更多的Follower赶上并重新进入ISR被Leader应答的消息都至少有min.insync.replicas个副本，因此能够容忍min.insync.replicas-1个副本同时宕机；

发送的acks=1和0消息会出现丢失情况，为不丢失消息可配置生产者acks=all & min.insync.replicas >= 2；

LEO&HW

LEO(log end offset) ，即⽇志末端偏移，指向了副本日志中下⼀条消息的位移值(即下一条消息的写⼊位置)；
HW(high watermark)，即已同步消息标识，因其类似于⽊桶效应中短板决定⽔位高度，故取名高水位线；所有高水位线以下消息都是已备份过的，消费者仅可消费各分区Leader⾼水位线以下的消息，对于任何⼀个副本对象而言其HW值不会大于LEO值；Leader的HW值由ISR中的所有备份的LEO最小值决定(Follower在发送FetchRequest时会在PartitionFetchInfo中会携带Follower的LEO)；

Kafka原本使⽤用HW来记录副本的备份进度，HW值的更新通常需要额外一轮FetchRequest才能完成，存在一些边缘案例导致备份数据丢失或导致多个备份间的数据不一致，Kafka新引入了Leader epoch解决HW 截断产⽣的数据一致性的问题；

故障恢复

Broker故障恢复

Kafka从0.8版本开始引⼊了一套Leader选举及失败恢复机制:⾸先需要在集群所有Broker中选出⼀个Controller，负责各Partition的Leader选举以及Replica的重新分配，当出现Leader故障后，Controller会将Leader/Follower的变动通知到需为此作出响应的Broker； Kafka使⽤ZooKeeper存储Broker、Topic等状态数据，Kafka集群中的Controller和Broker会在ZooKeeper指定节点上注册 Watcher(事件监听器器)，以便在特定事件触发时，由ZooKeeper将事件通知到对应Broker；

Broker故障场景分析

Broker与其他Broker断开连接

Broker0和其余Broker都断开了连接，由于ZooKeeper还能接收到Broker0的⼼跳，因此ZooKeeper认为Broker0依然存活；

Partition0

Broker0中的副本为Partition0的Leader，当Broker0超过replica.lag.time.max.ms没接收到Broker1、Broker2的FetchRequest请求后， Broker0选择将Partition0的ISR收缩到仅剩Broker0本身，并将ISR的变更更同步到ZooKeeper；Broker0需要根据min.insync.replicas的配置决定是否继续接受生产者数据； Partition1 超过replica.lag.time.max.ms后，Broker1会将Broker0中的副本从Partition1的ISR中移除，若后续Broker0恢复连接并赶上了Broker1，则Broker1还会再将Broker0重新加入Partition1的ISR；

当Broker发生故障后，由Controller负责选举受影响Partition的新Leader并通知到相关Broker：

当Broker出现故障与ZooKeeper断开连接后，该Broker在ZooKeeper对应的znode会自动被删除，ZooKeeper会触发Controller注册在该节点的Watcher；
Controller从ZooKeeper的/brokers/ids节点上获取宕机Broker上的所有Partition；
Controller再从ZooKeeper的/brokers/topics获取所有Partition当前的ISR；
对于宕机Broker是Leader的Partition，Controller从ISR中选择幸存的Broker作为新Leader；
最后Controller通过LeaderAndIsrRequest请求向的Broker发送LeaderAndISRRequest请求；

Broker与ZooKeeper断开连接

Broker0与ZooKeeper断开连接后，ZooKeeper会⾃自动删除该Broker对应节点，并且认为Broker0已经宕机，则对于

Partition0

ZooKeeper删除节点后，该节点上注册的Watcher会通知Controller，Controller会发现Broker0为Partition0的Leader，于是从当前存活的ISR中选择了了Broker2作为Partition0的新Leader。Controller通过LeaderAndIsrRequest将Leader变更更通知到Broker1、Broker2，于是Broker1改向Broker2发送Partition0数据的FetchRequest请求；生产者每隔60秒会从bootstrap.servers中的Broker获取最新的metadata，当发现Partition0的Leader发生变更更后，会改向新 Leader-Broker2发送Partition0数据；另一边Broker0收不到ZooKeeper通知，依然认为自己是Partition0的Leader;由于Broker1、 Broker2不再向Broker0发送FetchRequest请求，缺失了ISR应答的Broker0停止写入acks=all的消息，但可以继续写入acks=1的消息；在replica.lag.time.max.ms时间后，Broker0尝试向ZooKeeper发送ISR变更请求但失败了，于是不再接收生产者的消息；当Broker0与ZooKeeper恢复连接后，发现自己不不再是Partition0的Leader，于是将本地日志截断(为了保证和Leader数据一致性)，并开始向Broker2发送FetchRequest请求；在Broker0与ZooKeeper失联期间写⼊入Broker0的所有消息由于未在新Leader中备份，这些消息都丢失了；

Partition1

Broker0中的副本只是作为Partition1的Follower节点，⽽而Broker0与Broker1依然保持连接，因此Broker0依然会向Broker1发送 FetchRequest，只要Broker0能继续保持同步，Broker1也不不会向ZooKeeper变更更ISR；

Broker故障恢复过程

Broker发生故障后，由Controller负责选举受影响Partition的新Leader并通知到相关Broker：

当Broker出现故障与ZooKeeper断开连接后，该Broker在ZooKeeper对应的znode会自动被删除，ZooKeeper会触发Controller注册在该节点的Watcher；
Controller从ZooKeeper的/brokers/ids节点上获取宕机Broker上的所有Partition(简称set_p)；
Controller再从 ZooKeeper的/brokers/topics获取set_p中所有Partition当前的ISR;对于宕机Broker是Leader的Partition，Controller从ISR中选择幸存的Broker作为新Leader;最后Controller通过LeaderAndIsrRequest请求向set_p中的Broker发送LeaderAndISRRequest请求；
受到影响的Broker会收到Controller发送的LeaderAndIsrRequest请求后，Broker通过ReplicaManager的becomeLeaderOrFollower方法响应LeaderAndIsrRequest：新Leader会将HW更更新为它的LEO值，而Follower则通过一系列策略截断log以保证数据一致性；

Controller 故障场景分析

Controller与ZooKeeper断开连接

ZooKeeper会将Controller临时节点删除，并按照下节的故障恢复过程重新竞选出新Controller；而原本的Controller由于无连上ZooKeeper，它什么也执行不了；当它与ZooKeeper恢复连接后发现自己不再是Controller，会在Kafka集群中充当一个普通的 Broker；

Controller与某个Broker断开连接

因为Controller无法通知到Broker0，所以Broker0不晓得Partition0的Leader已经更换了，所以也会出现上述的短暂服务不可用并可能发生数据丢失；

Controller 故障恢复过程

集群中的Controller也会出现故障，因此Kafka让所有Broker都在ZooKeeper的Controller节点上注册一个Watcher；Controller发生故障时对应的Controller临时节点会自动删除，此时注册在其上的Watcher会被触发，所有活着的Broker都会去竞选成为新的 Controller(即创建新的Controller节点，由ZooKeeper保证只会有一个创建成功)；竞选成功者即为新的Controller，会在ZooKeeper的下述节点上注册Watcher，以监控各Broker运行状态、负责Leader宕机的失败恢复，并对管理脚本做出响应；

在/admin节点上注册Watcher，以应对管理理员脚本对Topic及Partition的影响；
在/brokers/ids节点上注册Watcher，以获取各Brokers的状态变化；
在/brokers/topics节点上注册Watcher，以监控每个Partition的ISR副本状态；

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2022-11-10，如有侵权请联系 cloudcommunity@tencent.com 删除

zookeeper

kafka

本文分享自对线JAVA面试微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

zookeeper

kafka

登录后参与评论

0 条评论

热度