首页 > 其他分享 >Kafka服务端的各种机制实现原理

Kafka服务端的各种机制实现原理

时间：2024-08-10 14:25:46浏览次数：16  
标签：partition -- Partition HW Kafka 原理 Leader 服务端 hsTopic

文章目录

zookeeper整体元数据

Zookeeper客户端工具： prettyZoo下载地址

我们知道kafka集群选举相关都是基于zookeeper来实现的，kafka注册进zookeeper中的数据目录如下图所示

在这里插入图片描述



上图中，最外层的一个方框都是kafka注册的元数据，如果想要kafka的这些元数据在一个统一的一级目录下，那么我们就可以在kafka的conf/server.properties中，指定zookeeper集群地址时，在每个ip地址后面统一加一个目录

在这里插入图片描述



在那么多的元数据中，比较重要的数据就是brokers目录和controller。其中brokers目录下记录了当前kafka集群下的所有broker存活节点；还有所有的topic信息，某个topic下partition分区的情况 Leader是在哪个Broker节点上，整个partition分区的复制集都在哪些节点上

而controller就是记录了kafka集群的多个Broker中，哪一个Broker充当controller角色

在这里插入图片描述



Controller信息如下

在这里插入图片描述



Kafka集群中的每个Broker启动后，都会往Zookeeper注册一个临时节点/brokers/ids/{broker.id}。如果Kafka的服务器非正常关机，这时Zookeeper上的临时节点不会注销，Kafka服务重启后，就有可能无法注册这个临时节点而报错

在这里插入图片描述



Controller Broker选举机制

Kafka的集群需要选举一个Controller Broker角色来管理整个集群的Broker和partition分区相关的信息，并负责将这些信息同步给集群内其他Broker。

Kafka集群下的Broker启动时，都会去往Zookeeper中创建一个Controller的临时节点，注册信息如下图右下角所示。Zookeeper保证只有一个Broker能创建成功。

在这里插入图片描述



创建controller临时节点的 Broker需要与Zookeeper保存长连接，定时发送心跳；允许其他Broker监听controller临时节点状态。

如果zookeeper长时间收不到当前Controller Broker的心跳，就会删除这个临时节点，同时会向该临时节点对应的所有监听器广播节点变化事件。其他Broker就有一次开始尝试创建Controller临时节点。

Controller Broker主要就是监听zookeeper下的/brokers/ids 、/brokers/topics这些节点，感知对应的增减变化，并同步给集群下的其他Broker。



Leader Partition选举机制

一个Topic下可以存在多个partition分区，而每个partition分区由可以通过添加备份组成一个复制集，一个Partition复制集中会选举出一个Leader多个Follower，LeaderPartition负责接收Client的读写请求，并将消息优先保存，再通知其他FollowerPartition来同步消息



专有名词概念：

  • AR: Assigned Repllicas。

    表示Kafka分区中副本集下的所有副本，包括不存活的partition。也就是下方命令中的Replicas这一列

  • ISR

    表示当前副本集中还存活的partition，能保持和Leader同步的Follower集合。如果Follower长时间没有向Leader发送通信请求(超时时间由replica.lag.time.max.ms参数设定，默认30S，那么这个Follower就会被移出ISR中。

  • OSR

    保存移出ISR的partition



我们在创建Topic时，可以通过--partitions来指定分区数，通过--replication-factor 来指定一个复制集下存在的partition数量。

# 创建一个hsTopic   其中共有4个partition分区，  每个分区下的复制集个数是3也就是一个Leader 两个Follower
[root@worker1 ~]# kafka-topics.sh --create --topic hsTopic --partitions 4 --replication-factor 3 --bootstrap-server localhost:9092
Created topic hsTopic.

# 查看所有的topic列表，就能在最后看到我们上方创建的hsTopic
[root@worker1 ~]# kafka-topics.sh --list --bootstrap-server localhost:9092
__consumer_offsets
__transaction_state
disTopic
hsTopic

# 通过使用--describe参数 查看某个topic的详细信息
[root@worker1 ~]# kafka-topics.sh --describe --topic hsTopic --bootstrap-server localhost:9092
Topic: hsTopic	TopicId: OZC7rONtR-ShGjwuq0xGpA	PartitionCount: 4	ReplicationFactor: 3	Configs: 
	Topic: hsTopic	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
	Topic: hsTopic	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
	Topic: hsTopic	Partition: 2	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
	Topic: hsTopic	Partition: 3	Leader: 3	Replicas: 3,2,1	Isr: 3,2,1

# 其中最前面的一列表示分区号，Leader表示当前分区复制集的LeaderPartition所在的BrokerId, 
# Replicas表示复制集所在的brokerid，Isr表示当前复制集下还存活的Partition所在的brokerId



我们再去查看Zookeeper中的数据，下图中的数据，和我们上方命令中查询出来是也是一一对应的

在这里插入图片描述



Kafka的Leader Partition选举机制非常高效，直接使用的是Replicas这个集合中的第一个作为的Leader，在选举Leader Partition时，会按照AR的排名顺序，靠前的优先选举，只要这个靠前的Partition还在ISO中(需要存活)，那么这个Partition就会被选举成Leader Partition。



测试，如下Partition2所示，它当前的Leader在broker2上，如果我此时把Broker2上的Kafka服务停止，我们来看看它的Leader 是否是选举到Broker3上去了

[root@worker1 ~]# kafka-topics.sh --describe --topic hsTopic --bootstrap-server localhost:9092
Topic: hsTopic	TopicId: OZC7rONtR-ShGjwuq0xGpA	PartitionCount: 4	ReplicationFactor: 3	Configs: 
	Topic: hsTopic	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
	Topic: hsTopic	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
	Topic: hsTopic	Partition: 2	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
	Topic: hsTopic	Partition: 3	Leader: 3	Replicas: 3,2,1	Isr: 3,2,1

# 停止Broker2上的服务
[root@worker2 ~]# kafka-server-stop.sh 
[root@worker2 ~]# jps
3313 Jps
1496 QuorumPeerMain



Zookeeper上的数据也同步更新了，没有了broker.id = 2 的节点了

在这里插入图片描述



我们再来查看topic的信息，可以发现partition2的Leader果然是重新选举到broker.id=3的服务器上去了，并且Isr这一列也没有了broker.id =2的信息了

[root@worker1 ~]# kafka-topics.sh --describe --topic hsTopic --bootstrap-server localhost:9092
Topic: hsTopic	TopicId: OZC7rONtR-ShGjwuq0xGpA	PartitionCount: 4	ReplicationFactor: 3	Configs: 
	Topic: hsTopic	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1
	Topic: hsTopic	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,3
	Topic: hsTopic	Partition: 2	Leader: 3	Replicas: 2,3,1	Isr: 3,1
	Topic: hsTopic	Partition: 3	Leader: 3	Replicas: 3,2,1	Isr: 3,1

在这里插入图片描述



Leader Partition自动平衡机制

Leader Partition比较繁忙，既要处理Client的请求，又要处理复制集的数据同步，Kafka会尽力将Leader Partition分布在不同的Broker上。

因为不可抗力因素导致Leader Partition重新选举之后，就会出现多个Leader Partition在一个Broker上，这个Broker的压力就会明显高于其他Broker。



Kafka存在一个Leader Partition自动平衡机制：Kafka认为AR中的第一个节点就应该是Leader，这是它认为最理想的情况。Controller Broker会每隔leader.imbalance.check.interval.seconds时间，检查所有的Broker，如果发现某个Broker中Leader Partition的不理想比例达到了leader.imbalance.per.broker.percentage 阀值，就会触发一次自动平衡。

这个机制涉及到conf/server.properties配置文件中的几个重要参数：

官网 服务端的参数说明

# 注意要修改集群中所有broker的文件，并且要重启Kafka服务才能生效。
#1 自平衡开关。默认true，值的类型是boolean
auto.leader.rebalance.enable
Type:	boolean
Default:	true

#2 自平衡扫描间隔
leader.imbalance.check.interval.seconds
Type:	long
Default:	300

#3 自平衡触发比例
leader.imbalance.per.broker.percentage
Type:	int
Default:	10



也可以通过手动通过kafka-leader-election.sh脚本触发一次自平衡机制

接着Leader Partition选举机制的案例，此时我重启了broker.id=2的服务，此时各个LeaderPartition的分布情况如下所示

# 其中的partition2 因为触发了重选选举，导致了此时的Leaderpartition并不是在理想的节点Broker2上
[root@worker1 ~]# kafka-topics.sh --describe --topic hsTopic --bootstrap-server localhost:9092
Topic: hsTopic	TopicId: OZC7rONtR-ShGjwuq0xGpA	PartitionCount: 4	ReplicationFactor: 3	Configs: 
	Topic: hsTopic	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
	Topic: hsTopic	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,3,2
	Topic: hsTopic	Partition: 2	Leader: 3	Replicas: 2,3,1	Isr: 3,1,2
	Topic: hsTopic	Partition: 3	Leader: 3	Replicas: 3,2,1	Isr: 3,1,2
	
	
# 手动触发所有Topic的Leader Partition自平衡 
# --election-type 选举的类型选择preferred，表示按照最理想的方式选leader  ,之后指定哪一个topic下的哪一个partition要进行自平衡
[root@worker1 ~]# kafka-leader-election.sh --election-type preferred --topic hsTopic --partition 2 --bootstrap-server localhost:9092
Successfully completed leader election (PREFERRED) for partitions hsTopic-2

# 此时 partition2的leader就成了broker2节点上的了，变为了最理想的情况
[root@worker1 ~]# kafka-topics.sh --describe --topic hsTopic --bootstrap-server localhost:9092
Topic: hsTopic	TopicId: OZC7rONtR-ShGjwuq0xGpA	PartitionCount: 4	ReplicationFactor: 3	Configs: 
	Topic: hsTopic	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
	Topic: hsTopic	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,3,2
	Topic: hsTopic	Partition: 2	Leader: 2	Replicas: 2,3,1	Isr: 3,1,2
	Topic: hsTopic	Partition: 3	Leader: 3	Replicas: 3,2,1	Isr: 3,1,2



注意：

在生产环境下不建议使用自平衡，因为这是一个比较耗时的操作，涉及到了大量消息的转移与同步，并且在这个过程中可能存在丢失消息的风险。

对于性能要求比较高的线上环境，会选择将参数auto.leader.rebalance.enable设置为false，关闭Kafka的Leader Partition自平衡操作，而用其他运维的方式，在业务不繁忙的时间段，手动进行Leader Partiton自平衡，尽量减少自平衡过程对业务的影响。



Partition故障恢复机制

Leader partition所在的broker服务停止后，Kafka就会触发partition重新选举，原来LeaderPartition中还未来得及同步的数据怎么办？



专有名词概念：

  • LEO(log end offset)，每个partition的最后一个offset

    Leader Partition接收Client的请求将消息持久化后就会更新LEO，Follower Partition从Leader Partition每同步一次消息就更新一次LEO

  • HW(High Watermark)，复制集下的最小的一个LEO

    Follower Partition向Leader Partition拉取消息，并上报LEO给leader，Leader Partition就会计算出HW，Follower下次拉取消息再更新本机的HW。HW之后的消息在复制集下的所有Partition都存在，所以可以认为这是安全的消息，允许消费者进行消费；而HW之前的消息为不安全消息，对消费者是不可见的。

在这里插入图片描述



也就是说，生产者新发送的消息并不会立刻被消费者消费，需要等到HW移动到该消息对应的offset之后消费者才会消费该消息。



Follower Partition出现故障的处理流程：

  • 当前故障Partition 移出ISR列表，其他Follower Partition正常同步消息
  • 当故障的Partition恢复后，不会立刻加入到ISR列表中，会读取到出现故障时本机保存的HW值，删除HW之前的消息，然后去往Leader同步消息
  • 当同步的消息也就是LEO达到当前复制集的HW后，加入到ISR列表中

在这里插入图片描述



Leader Partition出现故障的处理流程：

  • 从AR中，靠后一个重选选举出一个Leader Partition，此时消息可能还没有同步完成，新leader和LEO可能是小于老leader的LEO的。
  • 新leader也删除HW之前的消息
  • Kafka的消息只能以leader的备份为准，其他Follower partition将各自Log文件中高于HW的消息删除，并开始向新Leader partition开始同步消息
  • 当之前的Leader partition恢复之后，将作为Follower partition，先读取到本机的HW，将HW之前的消息全部删除。然后开始向当前Leader partition同步消息，当LEO达到复制集的HW时，加入到ISR列表中



在这里插入图片描述



在这个过程当中，Kafka注重的是保护多个副本之间的数据一致性。但是这样，消息的安全性就得不到保障。例如在上述示例中，原本Partition0中的4，5，6，7号消息就被丢失掉了。



在这个机制中，有一个很重要的前提，那就是各个broker中记录的HW是一致的，可是HW和LEO同样是一个分布式的值，怎么保证HW在多个Broker中是一致的

Leader将当前最新计算出的HW同步给其他Follower这一过程中也存在一个延时，某些Follower先拉取消息就得到了最新的HW，某些Follower还没有拉取消息。如果在这个过程中，Leader所在Broker节点宕机了，这不就造成了前一部分Follower是最新的刚同步的HW，而后一部分Follower还没有得到最新的HW，那此时该怎么办嘞？



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

HW的值在一个partition复制集中并不是总是一致的。



不一致的产生的原因：

Follower partition先向loader partition拉取消息，才能向leader上报LEO值。leader得到所有Follower上报的LEO之后计算出HW。Follower 下次拉取消息才能一起拉取最最新的HW，再更新HW。Leader和Follower更新HW的值就存在一定的时间延迟，如果有多个Follower，他们之间保存的HW也存在一定时间的不统一。

此时如果服务一切正常，那么Leader还是会稳步的向前推进HW；但是当Leader partition发生了切换，所有的Follower partition都按照自己的HW进行数据恢复，就会出现数据不一致的情况。

在这里插入图片描述



因此，Kafka还设计了Epoch机制，来保证HW的一致性。

  • epoch是一个递增的版本号，当发生Leader partition切换时就会递增，之间低版本的epoch就会无效

  • 每个Leader在上任之初都会新增一个epoch。这个记录包含新增的epoch版本号和当前Leader partition最新写入的第一个消息的偏移量offset。例如(25,100)，25表示epoch的版本号，100表示当前新Leader写入的第一条消息offset是100。

    Broker会将这个epoch数据保存到内存中，并且会持久化到本地一个leader-epoch-checkpoint文件当中。

  • 这个leader-epoch-checkpoint文件本来就会在整个副本集中同步。当出现了Leader partition切换，新Leader就会进行更新添加自己的epoch

  • 接下里其他Follower partition要更新数据时，就可以不再依靠自己本机的HW值判断拉取消息的起点，而可以根据这个最新的epoch条目来判断



这个关键的leader-epoch-checkpoint文件保存在Broker上每个partition对应的本地目录中。这是一个文本文件，可以直接查看。他的内容大概是这样样子的

[oper@worker1 disTopic-0]$ cat leader-epoch-checkpoint 
0
1
29 2485991681



其中第一行版本号，第二行表示下面的记录数。这两行数据没有太多的实际意义。

从第三行开始，可以看到两个数字。这两个数字就是epoch 和 offset。epoch就是表示leader的epoch版本。从0开始，当leader变更一次epoch就会+1。

offset则对应该epoch版本的leader写入第一条消息的offset。可以理解为用户可以消费到的最早的消息offset。

标签：partition,--,Partition,HW,Kafka,原理,Leader,服务端,hsTopic
From： https://blog.csdn.net/qq_44027353/article/details/141062711

相关文章

  • ConcurrentHashMap的原理
    背景我们知道hashmap是一个线程不安全的数据结构，在多线程编程的时候，多个线程同时向hashmap中put元素的时候，会发生数据丢失。多线程put操作后，再get操作导致死循环。多线程put非NULL元素后，get操作得到NULL值。使用为了保证并发安全，我们使用hashmap的时候，建议是使用ConcurrentHas......
  • 编译原理
     Java程序员或者是Java工程师逐渐成长为高级开发的时候，他看问题的方式写工作的方式都有自己的一套方式。每个人的工作效率不一样，但是都要达到同一个目的。对于开发任务而言，测试通过是最低的要求。Java的集成开发环境十分优秀，西方美国的Eclipse到欧洲的Idea，程序员开......
  • Kafka整合SpringBoot
    前文Kafka客户端详解引入依赖<properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target>......
  • CORS与JSONP漏洞原理及其复现
    文章目录CORS跨域共享介绍同源策略配置产生条件复现自动化爆破工具JSONP跨域回调介绍复现自动审查工具安装教程使用教程在现代Web开发中，跨域资源共享（CORS）和JSONP是两种重要的跨域通信技术。CORS通过设置HTTP头来控制资源共享，而JSONP则利用了本文将深入探讨......
  • spring原理（第十三天）
    WEBRequestMappingHandlerMapping与RequestMappingHandlerAdapter俩是一对，分别用来处理@RequestMapping映射调用控制器方法、并处理方法参数与方法返回值DispatcherServlet初始化DispatcherServlet是在第一次被访问时执行初始化,也可以通过配置修改为Tomcat......
  • 【数据结构】关于栈你必须知道的内部原理！！！
    前言：......
  • 回归预测|一种多输入多输出的粒子群优化支持向量机数据回归预测Matlab程序PSO-MSVR非f
    回归预测|一种多输入多输出的粒子群优化支持向量机数据回归预测Matlab程序PSO-MSVR非for循环实现原理上进行修改多输出文章目录前言回归预测|一种多输入多输出的粒子群优化支持向量机数据回归预测Matlab程序PSO-MSVR非for循环实现原理上进行修改多输出一、PSO-MSVR......
  • 京东面试：说说CMS工作原理？
    CMS（ConcurrentMarkSweep）垃圾收集器是以“最短的停顿”著称的垃圾回收器，因此也是JDK9之前使用最广泛的垃圾回收器之一。那么，问题来了，为什么CMS能实现最短停顿时间？CMS垃圾回收器的工作原理又是啥呢？接下来，我们一起来看。CMS工作原理CMS之所以能实现最短停顿时间是和它的......
  • React原理之React整体渲染流程
    前置知识：深度优先搜索(DFS)、Fiber节点在上一篇React原理篇之React整体架构解读中，提到了Fiber架构中的几个核心概念：Scheduler（调度器）：根据任务的优先级安排任务执行顺序。Reconciler（协调器）：根据新旧虚拟DOM树的差异确定需要更新的部分。Renderer（渲染器）：将更新的虚拟......
  • 相机成像原理
    https://mp.weixin.qq.com/s?__biz=MzU2NjU3OTc5NA==&mid=2247545276&idx=2&sn=458b782fadf8dda3b14100a375d82032&chksm=fca83681cbdfbf9782cdd89d3337bbf13c50ebccd711ec4ec5dc9b1ab7b5b3b7472d884ee044&scene=271.空间三维坐标系三维空间坐标系即世界坐标系，是一个绝对的坐标......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99