Kafka-集群架构设计

发布时间:2023年12月18日

Kafka的Zookeeper元数据梳理

zookeeper整体数据

Kafka将状态信息保存在Zookeeper中,这些状态信息记录了每个Kafka的Broker服务与另外的Broker服务 有什么不同。通过这些差异化的功能,共同体现出集群化的业务能力。这些数据,需要在集群中各个Broker 之间达成共识,因此,需要存储在一个所有集群都能共同访问的第三方存储中。

Kafka的整体集群结构:

Kafka的集群中,最为主要的状态信息有两个:

  • 在多个Broker中,需要选举出一个Broker,担任 Controller角色。由Controller角色来管理整个集群中的分区和副本状态。

  • 在同一个Topic下的多个 Partition中,需要选举出一个Leader角色。由Leader角色的Partition来负责与客户端进行数据交互。

Zookeeper数据整体分布图:

Zookeeper客户端工具: prettyZoo。下载地址:Releases · vran-dev/PrettyZoo · GitHub

Controller Broker选举机制

在Kafka集群进行工作之前,需要选举出一个Broker来担任Controller角色,负责整体管理集群内的分区和副本状态。选举Controller的过程就是通过抢占Zookeeper的/controller节点来实现的。

当一个集群内的Kafka服务启动时,就会尝试往Zookeeper上创建一个/controller临时节点,并将自己的 brokerid写入这个节点。而Zookeeper会保证在一个集群中,只会有一个broker能够成功创建这个节点。这个注册成功的broker就成 了集群当中的Controller节点。节点内容:

{"version":2,"brokerid":0,"timestamp":"1702350416126","kraftControllerEpoch":-1}

临时节点:长时间断开连接,会被自动删除

监听机制:监听节点的状态,节点被删除后给所有监听器广播节点状态变化事件

Controller节点,负责监听Zookeeper中的其他一些关键节点,触发集群的相关管理工作:

  • 监听Zookeeper中的/brokers/ids节点,感知Broker增减变化。

  • 监听/brokers/topics,感知topic以及对应的partition的增减变化。

  • 监听/admin/delete_topic节点,处理删除topic的动作。

另外,Controller还需要负责将元数据推送给其他Broker。

Leader Partition选举机制

在Kafka中,一个Topic下的所有消息,是分开存储在不同的Partition中的。

使用kafka-topics.sh脚本创 建Topic时

  • 通过--partitions参数指定Topic下包含多少个Partition

  • 通过--replication-factors参数指定每个Partition有几个备份

在一个Partition的众多备份中,需要选举出一个Leader Partition,负责对接所有的客户端请求,并将消息优先保存,然后再通知其他Follower Partition来同步消息。

涉及的基础概念:

  • AR: Assigned Repllicas。 表示Kafka分区中的所有副本(存活的和不存活的)

  • ISR: 表示在所有AR中,服务正常且保持与Leader同步的Follower集合。如果Follower长时间没有向Leader发送通信请求(超时时间由replica.lag.time.max.ms参数设定,默认30S),那么这个Follower就会被踢出ISR。(在老版本的Kafka中,还会考虑Partition与Leader Partition之间同步的消息差值,大于参数replica.lag.max.messages条就会被移除ISR。现在版本已经移除了这个参数)

  • OSR:表示从ISR中踢出的节点。记录那些服务有问题,延迟过多的副本。

# 查看状态命令
bin/kafka-topics.sh -bootstrap-server localhost:9092 --describe --topic disTopic

在选举Leader Partition时,会按照AR中的排名顺序,靠前的优先选举。只要当前Partition在ISR列表中,也就是是存活的,那么这个节点就会被选举成为Leader Partition。(AR就是Replicas列中的Broker集合)

Leader Partition自动平衡机制

默认情况下,Kafka会尽量将Leader Partition分配到不同的Broker节点上,用以保证整个集群的性能压力能够比较平均。但是,经过Leader Partition选举后,这种平衡就有可能会被打破,让Leader Partition过多的集中到同一个Broker上,影响集群的整体性能。为此,Kafka设计了Leader Partition自动平衡机制,当发现Leader分配不均衡时,自动进行Leader Partition调整。

Leader Partition自平衡逻辑:会认为AR当中的第一个节点就应该是Leader节点。这种选举结果称为preferred election 理想选举结果。Controller会定期检测集群的Partition平衡情况,在开始检测时,Controller会依次检查所有的Broker。当发现这个Broker上的不平衡的Partition比例高于leader.imbalance.per.broker.percentage阈值时,就会触发一次Leader Partiton的自平衡。

官方文档的部分截图:

涉及Broker中server.properties配置文件的几个重要参数:

#1 自平衡开关。默认true
auto.leader.rebalance.enable
Enables auto leader balancing. A background thread checks the distribution of partition leaders at regular intervals, configurable by `leader.imbalance.check.interval.seconds`. If the leader imbalance exceeds `leader.imbalance.per.broker.percentage`, leader rebalance to the preferred leader for partitions is triggered.
Type:   boolean
Default:    true
Valid Values:   
Importance: high
Update Mode:    read-only
?
#2 自平衡扫描间隔
leader.imbalance.check.interval.seconds
The frequency with which the partition rebalance check is triggered by the controller
Type:   long
Default:    300
Valid Values:   [1,...]
Importance: high
Update Mode:    read-only
?
#3 自平衡触发比例
leader.imbalance.per.broker.percentage
The ratio of leader imbalance allowed per broker. The controller would trigger a leader balance if it goes above this value per broker. The value is specified in percentage.
Type:   int
Default:    10
Valid Values:   
Importance: high
Update Mode:    read-only

注意:使用集群时,要修改集群中所有broker的文件,并且重启Kafka服务才能生效

# 手动触发所有Topic的Leader Partition自平衡
bin/kafka-leader-election.sh --bootstrap-server worker1:9092 --election-type preferred ?--topic secondTopic --partition 2

Leader Partition自平衡的过程是一个非常重的操作,因为要涉及到大量消息的转移与同步。并且,在这个过程中,会有丢消息的可能。所以在很多对性能要求比较高的线上环境,会选择将参数auto.leader.rebalance.enable设置为false,关闭Kafka的Leader Partition自平衡操作,用其他运维的方式,在业务不繁忙的时间段,手动进行Leader Partiton自平衡,尽量减少自平衡过程对业务的影响。

Partition故障恢复机制

当一组Partition中选举出了一个Leader节点后,这个Leader节点就会优先写入并保存Producer传递过来的消息,然后再同步给其他Follower。当Leader Partition所在的Broker服务发生宕机时,Kafka就会触发Leader Partition的重新选举。

Kafka为了消息能够在多个Parititon中保持数据同步,内部记录了两个关键数据:

  • LEO(Log End Offset): 每个Partition的最后一个Offset

每个Partition都会记录自己保存的消息偏移量。leader partition收到并记录了生产者发送的一条消息,就将LEO加1。follower partition从leader partition同步消息,每同步到一个消息,自己的LEO就加1。通过LEO值,就知道各个follower partition与leader partition之间的消息差距。

  • HW(High Watermark): 一组Partiton中最小的LEO。

follower partition每次往leader partition同步消息时,都会同步自己的LEO给leader partition。这样leader partition就可以计算出这个HW值,并最终会同步给各个follower partition。leader partition认为这个HW值以前的消息,都是在所有follower partition之间完成了同步的,是安全的。这些安全的消息可以被消费者拉取。而HW值之后的消息,是不安全的,可能丢失的。这些消息如果被消费者拉取消费了,就有可能造成数据不一致。

当服务出现故障时,如果是Follower发生故障,不会影响消息写入,只是少了一个备份而已。Kafka会做如下处理:

1、将故障的Follower节点临时踢出ISR集合。其他Leader和Follower继续正常接收消息。

2、出现故障的Follower节点恢复后,不会立即加入ISR集合。该Follower节点会读取本地记录的上一次的HW,将自己的日志中高于HW的部分信息全部删除掉,然后从HW开始,向Leader进行消息同步。

3、等到该Follower的LEO大于等于整个Partiton的HW后,就重新加入到ISR集合中。也就是说这个Follower的消息进度追上了Leader。

如果是Leader节点出现故障,Kafka为了保证消息的一致性,会做如下处理:

1、Leader发生故障,会从ISR中进行选举,将一个原本是Follower的Partition提升为新的Leader。这时,消息有可能没有完成同步,所以新的Leader的LEO会低于之前Leader的LEO。

2、Kafka中的消息都只能以Leader中的备份为准。其他Follower会将各自Log文件中高于HW的部分全部清理掉,然后从新的Leader中同步数据。

3、旧的Leader恢复后,将作为Follower节点,进行数据恢复。

在这个过程当中,Kafka注重的是保护多个副本之间的数据一致性。但是消息的安全性就得不到保障。

思考:HW和LEO是一个分布式的值,怎么保证HW在多个Broker中是一致的呢?

HW一致性保障-Epoch更新机制

文章来源:https://blog.csdn.net/weixin_58482311/article/details/134983648
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。