我们接下来了解一下leader和follower
在Kafka 0.8以前的版本
它并没有提供HA（高可用）机制
没有HA其实就是说数据
只有一个副本 一旦挂了就没有了
Broker你总归是会遇到宕机的
那宕机机器上
所有的partition无法继续提供服务
如果这个broker永远无法再恢复的话
或者我们遇到的磁盘故障
这些数据将会彻底丢失
因为没有副本
所以在0.8以前的版本
一旦一个 broker挂
它的影响是非常大
所以它不满足Kafka的数据持久性
以及它的所谓的叫delivery guarantee的设计目标
跟这个目标是相悖的
同时producer也不能再将
这些数据存在于这些partition中
我们再看一下
它以前的这种重试能不能解决这个问题
如果producer使用的是同步模式
producer它会尝试重新发送做重试
因为你之前写的数据
你写的过程中正好丢了
你重试的时候
遇到三次错误以后它可能就会抛异常
用户可以停止发送
也可以继续发送
前者可能会导致数据的一个阻塞
后者可能会导致
发往该broker的数据丢失
这个其实是不好的
你使用异步模式的话
也是会重试n次以后会抛异常
但是它只会把异常记录起来
然后你可以继续发送数据
这会导致数据丢失
并且用户只能通过日志发现这个问题
也就是说Kafka的producer端
它其实是分同步和异步的
注意啊
异步是不能保证数据一定传达到对端的
就一定保证它传到broker的
同步的话相对来说会好一些
也就是说在没有副本的情况下
一旦机器宕机或者说某个broker停止工作
则会导致整个系统的可用性降低
因为你是这样的
其实如果你只有一个副本
如果你的消费速度非常快
可能影响的面比较小
但是一旦你的消费速度是存在延迟
你之前发的数据可能还没来得及消费 那就丢了
另外就是说你producer端也发不出去了
因为你同步发要么抛失败
你要么继续无限重试等待
或者说重试失败
抛出异常以后你继续发送
它可能卡住
因为这个partition没人负责了
你写不进去
或者是说如果你异步发送的话
因为你只是在异步后台发送
出问题的时候
也是你只能记个日志
然后你也发现不了这个问题
所以说没有副本
过去的数据可能丢了
你正在发的可能会阻塞
所以对整个系统的可用性影响还是非常大的
引入副本以后
同一个partition说白了就是有多个副本
两副本 三副本
这个时候我们通常做法就是说
在几个副本之间选一个leader
producer和consumer
只和leader做交互
这样的话
其它的replica我们叫follower
其它的follower再从leader
去同步数据复制数据
因为你需要保证同一个partition
多个replica之间的数据一致性
也就是说一个宕机后其它的replica
能够继续提供服务
并且呢不会造成数据重复
也不能造成数据丢失
这个时候你如果没有一个leader
所有replica同时进行一个读写
我们知道类似像Paxos多个replica之间互相同步数据
数据一致性和有序性其实是很难保证
尤其是有序性是比较难保证的
也大大增加了replica复制的实现的复杂性
所以包括像为什么Raft的也是一样选leader
选leader的方式
它从工程实现上会比较简单一点
所以Kafka它也是一样的
就会引入一个leader
leader负责数据的读写
也就是说你的follower producer
只和leader的进行交互
follower向leader干嘛
顺序的去fetch数据
这样的话你整个系统设计会变得非常简单
Kafka集群是依赖ZooKeeper
选所谓的partition leader的
因为每一个partition可能都有三个副本
10个partition的话每个人都有三个副本
其实就有30个partition需要去处理
最简单的方案就是依赖ZooKeeper
来为每一个partition来去选一个什么
选一个leader
大家比较容易想到这个做法嘛
比方说我们在这个场景里面
我的所有的follower都在
ZooKeeper上设置一个watch
我们知道ZooKeeper有watch机制
一旦leader宕机
对应的这种临时节点就删除了
所有此时所有的follower
肯定会收到一个event
就是说那个节点被删除掉了
收到一个这样的事件 这时候follower
大家都会尝试去创建该节点
也就是说我再去创建一个临时节点
创建成功的人
它就会成为新的leader
其它replica就变成了follower
其实说白了就是利用ZooKeeper
去创建某一个路径
然后来看谁抢成功就做leader
这样的方案
以前的版本是这么做的
但是它有个缺陷
首先也是由ZooKeeper特性导致的
因为ZooKeeper虽然能够保证watch按顺序触发
但是它并不能保证
同一时刻所有replica就是
所有的follower或者所有的副本
看到的状态是一致
因为它顺序触发
然后你某一个时刻要通知下来
通知到每个节点
导致大家看到的状态不一致
所以它看到的状态是不一致的
所以有可能出现脑裂 这是第一个
第二个叫羊群效应 我们可以想象一下
如果宕机的broker上partition数量超级多
是不是意味着导致会触发多个watch
因为所有follower会被唤醒
导致集群的大量的调整
这个其实是叫羊群
一个broker上的partition太多
然后导致watch频繁的触发和重新订阅
这个也是不好
另外就是ZooKeeper的负载过重
因为每个replica都要为ZooKeeper
注册一个watch就是事件
你集群规模增加到几千个partition的时候
ZooKeeper就不堪重负
这个其实是它的一个缺陷
既然你每一个partition都要选一个leader
全部用那个ZooKeeper来做的话
整个watch是很恐怖夸张的
Kafka在leader的选举上做了一些小的优化
它在所有的broker中
刚刚我们之前讲consumer的时候也提到过
consumer group其实是有
所谓的group coordinator
就是所谓的组协调者
同样的在这里也是挑一个broker
选出一个controller 叫Kafka的controller
Kafka的controller
所有的partition leader都由controller来决定
controller会将leader的改变
直接通过RPC的方式
也就是比之前ZooKeeper这种
queue的方式通知更高效一些
它来负责RPC的通知
它通过RPC的方式来通知
就是需要为此做出响应的一个broker
简单看一下它的controller的一个选举
controller选举可能是需要依赖ZooKeeper的
最新版的Kafka彻底不依赖ZooKeeper了
但是这个版本比较新
可能稳定性还要再考验一下
我们这边其实还是讲的主流的版本
我们先看一下
首先每一个broker都会
在controller path
就这么一个路径上注册一个watch
因为你的broker的节点数量
比partition数量还是要少得多的
我们知道比方说你有10个broker
你就注册10个watch
但是你的partition是多少
你的partition虽然是10个partition
但是你注意你有follower的存在
你有三个 所以它乘30
所以扩大倍数就多了
好 broker来注册 controller path
它整个羊群效应就没有那么猛
当前的controller失败的时候
对应的controller path会自动消失
因为它是ZooKeeper的一个临时节点
这个时候watch会被触发
所有活着的broker都会去
竞选成为新的broker
因为都会尝试去创建的
controller path它也是个临时节点
最终其实只有一个能竞选成功
这个是由ZooKeeper来保障的
竞选成功就会成为新的leader
竞选失败的
重新在新的controller path上来注册watch
因为ZooKeeper的watch是一次性的
所以说它触发完之后它要重新去注册
这样的话就选出了一个leader
Kafka的partition leader的选举
选出了controller以后
我们再来看controller
它怎么去选Kafka的partition leader
第一 它从ZooKeeper中读取当前分区的
所有ISR叫in-sync replicas的一个集合
简单来说就是Kafka
因为它的副本跟我们的Paxos或者Raft
稍微有些区别就是说假设我是三副本
大家如果同步速度都很快
我们把这种同步速度非常快的
我们叫做ISR叫in-sync replicas叫同步副本
但是如果有一些节点
它可能因为一些异常或者故障
它同步速度特别的慢
如果我每次都等它写完我的producer才返回
肯定慢的不行
我就把它踢掉
把它从ISR里面踢掉
它离群了 你特别慢 我把你干掉
所以这个叫ISR
ZooKeeper肯定是尝试先读取所有
可用的ISR的副本的集合
调用配置的分区
选择算法选一个leader出来
它是一个什么样一个分区算法
我们看一下Kafka的partition的
replication它默认是自动分配的
我们先看看它是怎么实现的
在Kafka集群中
每个broker都有均等分配
partition leader的一个机会
在上个图里面我们看一下
这个箭头指向的是副本
broker 1中的partition 0为leader
就这个地方为leader
它的箭头指向的是什么
指向的是broker 2的
partition 0表示的它是个副本
同理broker 2的
partition 1这个是leader
它指向的broker 3的partition 1
是follower这个就是它的副本
所以我们看上图
每个broker按照broker
id 1234
依次分配出partition
下一个 broker为副本
然后如此循环迭代
副本都遵循这样一个原则
这个地方如果是leader 它挨着的
broker 2就是的follower
如果它是leader的话
我们看它就跳过来回到 broker 1
成为一个follower
好 所以它的分配算法相当于把n个broker
和待分配的i个partition排序
就先把这两个排序
排完序以后将i个partition分配到
i mod n个broker上
也就是说比方说
我是有四个partition 两个副本
创建了一个topic
它包含了四个partition两个application
我的partition 0怎么分配我们看一下
比方说partition 0这里是6
0 mod 6
它就命中第零个节点
就是broker 1就命中它上面
它的follower其实就是邻近的加一就好了
所以我们看到它第i个partition的第j个副本
分配在(i+j) mod n
这个j就是第一个第二个第三个
所以说它是顺序的第一个
第二个第三个
有点像一个负载均衡的策略
这个策略虽然看上去大家有均等的机会
但是从实际Kafka的实战来说
这种分配是过于理想化的
因为partition之间的数据或者说topic之间的数据
会导致某些broker的负载更重一些
这一块其实我们也魔改了Kafka的一些代码
在partition的leader的分配上
我们可能还要考虑到
有没有存在磁盘热点
有没有存在一个broker的热点
然后我们会重新去进行一个分配
当然了这个是事前的
运行中也有可能会某些
broker存在热点 某些partition存在热点
就某些磁盘存热点
可能它是要经常去挪一挪位置的
所以说这个是一个默认策略
回到我们刚刚说的controller
其实从ISR读出来就是刚刚前面的顺序列表
它会选第一个partition作为leader
这个就是controller和leader的一个做法
大部分系统一样就是Kafka处理
一个broker明确是否活着
它其实包含了几个条件
第一个条件就是说副本所在的节点需要
和ZooKeeper保持一个session
通过ZooKeeper的heartbeat来保障
就是我知道broker是否活着
第二个就是说从副本的最后一条offset与主副本
因为我们有主和follower
与主副本最后一条offset差值不超过一定的值
如果它们的值
两个offset 两个消息同步的速度差别特别大
我们就认为它是什么
离群的
新版本的Kafka 0.10.0把这个配置给干掉了
就是replica.lag.max.messages
因为这个不好配
所以新的它其实是通过时间来设定的
就是你们两个消息之间差了多少时间
你message这个是不好配的
所以后来又改成了用时间
我们刚刚说了leader会跟踪其同步
保持同步的replica列表
我们把这个列表叫做in-sync replica
就是我每次做数据分发复制的时候
其实我是优先给ISR列表的这些数据
如果一个follower宕机或者落后太多
把它从ISR中移除
当它满足以上的条件
它又会重新加入到ISR的集合里面
也就是说给你机会你继续同步数据
但如果你追上我来的时候
把你变成一个ISR的列表
ISR的引入主要是解决同步副本
或者说异步复制两种方案各自的缺陷
同步复制
有个副本宕机或者超时
就会拖慢整个副本的整组性能
有一个人特别慢
你肯定会拖慢整组性能
如果是仅仅用异步复制的话
当所有副本均落后于主副本的时候
一旦主副本挂掉
我要重新选举
又会存在消息丢失
当然了Kafka那个时候做的时候
我不知道为啥基于什么考虑
没有直接使用Raft
我觉得我看了一下最新版本的
Kafka的一些规划 其实也提到了
后续的话可以直接使用Raft
对于这种replicated log所谓的分布式日志系统
所有使用这种副本的一个多副本的一个策略
基本上主流大家都知道
要么就是基于Raft的
多数节点的ACK
节点一般称为leader和follower
Kafka将要使用 注意这个是将要使用
现有的机制叫pacificA的一个机制
就是基于所有节点ACK
注意Raft的是多数节点的ACK
但是pacificA是基于所有节点的ACK
一般把节点称为primary和secondary
这个是Kafka正在使用的
但是所有节点的ACK
我们待会还会提一下 它有个小的优化
还有bookkeeper是基于法定个数节点的ACK
一般称为writer和bookie
Pulsar就是Kafka的竞品正在使用
Kafka是在ZooKeeper中
动态维护了这么一个ISR的列表
这个ISR里的所有replica
都是能跟上leader的
它是取决于这个参数你delay多少时间
在这个时间内的话我认为你都是跟得上的
只有ISR的成员才能被选为leader
也就是说你跟得上我
你才可能有机会成为leader的
在这种模式下
对于f+1个replica
一个partition能在保证自己不丢失
已经commit消息的前提下
容忍f个replica的失败
注意像Raft的它是多数派
但是对于Kafka来说
它允许你失败f个replica
只要有一个副本成功它都能去写
对 所以在大多数使用场景中
这种模式是非常有利的 为什么
因为省机器
事实上为了容忍f个replica的失败
所谓的就是majority vote和ISR
在commit前需要等待的replica数量是一样的
为了容忍f个replica的失败
这个是一样的
但是ISR这种机制它需要总的replica的个数
几乎是majority vote的一半
假设我能允许f个失败
也就是说白了就是只有单节点
它需要的机器少得多
当然这种方式的的确确是数据存在丢失风险的
因为你只有一个副本挂了就没了
然后Kafka还有好多参数去配合
长期来看我觉得Kafka用Raft
可能还是会好的多一些
我们再来讲一讲
大家我们知道Kafka有leader以后
我们再来看看它follower了
它们之间的数据是怎么同步的
这就要涉及到Kafka的watermark
就是水位
watermark一般就叫水位线
还有一个叫log end offset
我们叫LEO
每个Kafka的副本对象都有两个重要的属性
一个叫LEO 一个叫HW
HW就对应的英文就是在上面标题
注意是所有的副本不是只有leader有
LEO叫日志末端位移
log end offset
记录的是什么
该副本底层日志
也就是说我的Kafka的消息
它下一条消息的位移值
注意是下一条消息
举个例子
也就是说如果LEO等于10
表示该副本保存了10条消息
位移范围就是0~9
水位值 HW表示的是什么
对于同一个副本对象而言
水位值不会大于LEO值
什么意思呢
也就是说我优先写了一条日志
我是更新日志末端位移
因为我数据写成功了
但是这个水位值是干嘛的呢
小于等于这个水位值的所有消息
被认为是已备份的
说白了就是告诉你
这个消息被复制到其它的副本
其实它副本都确认它们都复制成功了
大家可以理解吧
假设我有三副本
我的主副本写到第10条消息了
我的从副本可能才写到第8条消息
那是不是我的high watermark=8
表示我已经同步到8这个位置了
那是后面等于10
还有两条消息没被复制到
所以我们看一下图
high watermark前面的表示的已经被commit
已经被复制了
uncommitted表示我leader的最新增加的
但是还没来得及同步的
这个是一个很有意思的点
我们再来看一下这几个值
它分别是怎么玩怎么更新的
初始化的时候
leader的follower
HW和LEO都等于0
leader初始化的时候肯定都等于0
没有任何数据
leader的remote LEO指的是什么
这里面注意有这两个值
在leader它多了一个值
有一个叫remote LEO
注意是在leader中保存的follower的LEO
在leader端还要保存follower
你们写到哪来了
初始化也是0 说白了
除了我要保存自己的LEO
在leader它还要保证别人的
我得看看你写到哪来的
我们在后面再看看它有啥用
这个时候producer没有发送任何消息给leader
而follower它已经开始
不断地向leader去发送fetch请求
因为它要同步数据
但因为没有任何数据可以同步
所以说不值一提
但是 follower发送过来的fetch请求
你来发请求
我不能无限轮询
我们之前讲consumer其实也是类似的机制
它不能无限轮询
它会暂存到leader端的一个对象里面
就是它相当于你来请求我
但是我用长轮询的方式来对待你
目前是可以500毫秒
然后超时后再强制完成
暂存期间 如果producer的发送来数据
Kafka会自动唤醒该fetch请求
说白了它就是一个典型的长轮询
我们再记一下这个图
一开始是这样一个状态
注意follower是只有这两个值
HW和LEO这两个值
我们再来看一下
这个是初始阶段
这个时候我的producer
产生了一个消息
注意这个时候我的producer
我的leader是不是把LEO要更新成1了
因为它产生一个消息
这个时候是这种情况
我们再看下一个阶段
下一个阶段其实就是follower
要开始发送一个fetch请求了
因为待了500毫秒啥也没取到
下一个轮次我又来同步数据
这个时候因为你写了条消息
我们一开始producer往leader写消息的时候
其实它第一步就是说
把这个消息落地写到我的磁盘上
写到log文件
然后我的leader把 LEO更新成1
这个时候其实还会做一件事情
这个leader会尝试更新
HW high watermark值
它会尝试更新这个值
但是它发现更新不了 为什么
它发现你的remote LEO明明就还没有同步
我为啥要更新
我肯定不会去更新
所以说它发现它自己LEO=1的
但是它remote LEO还等于0
我肯定是取一个最小的作为我的HW值
结果发现你还没有更新
我肯定是不更新的
我们再来看一下
这个时候 follower来发fetch请求
它第一次来发出fetch请求的时候
肯定会把自己的offset给带过来
这个时候带什么值呢
follower会把自己的LEO值
offest这个值给请求带上来
带上来之后发现leader已经有一条数据了
然后它会response把这个数据带回来
我们带回来之后肯定就可以落盘了
这个时候我再把follower的 LEO=1
注意在带回来数据的时候
它response还会带一个值
把leader的 high watermark
这个值也顺便带回来
这个值带回来干嘛用
其实也是为了更新
HW的 我们后面再看它怎么更新的
我们再回溯一下本例中
follower发送fetch请求
它干了些啥事情
首先leader端
你带了一个fetch offset=0的请求过来
首先是leader先读取底层的log数据
然后更新remote LEO=0
为什么是0
因为此时follower还没有写入数据
你只是说你要开始从0开始同步
你还没写 肯定是更新remote LEO还是等于0
leader如何确认follower还未写入
因为说白了在你带的
fetch offset参数来决定
你同步到哪来了
你传的参数等于1
说明你已经有了
你把offset等于1带过来
请求fetch接口调我
调leader 我leader自然就知道
原来你写到1了
这个时候更新remote LEO=1
因为你带0我肯定是更新0的
leader还会做个事情
尝试更新分区的high watermark
因为这个时候leader LEO=1
remote LEO=0
然后HW值等于多少
我们说了之前是取leader和remote
取最小的 得到0
结果还是啥也没更新
然后把数据和当前分区的high watermark
还是等于0发送给follower副本
我要回复它
而follower收到fetch response以后
它会做几个事情
肯定是先落盘把message落盘
同时更新follower LEO
然后再更新follower HW
比较本地的LEO和当前的leader的
high watermark取最小值
和前面leader的更新high watermark值个步骤类似
发现还是等于0
故follower的high watermark等于0
这个时候第一轮的fetch的RPC结束了
我们会发现leader和follower
虽然都已经在log中保存的消息
但是分区的high watermark并没有更新
就是说我的HW还是等于0
实际上它是在第二轮fetch RPC的时候更新的
我们看一下
现在我们处于刚刚第二阶段是这样的
HW=0 remote LEO=0 LEO=1
这个是HW=0 LEO=1
我们目前处于这个状态
好 我们看下一个状态
下一个状态调用了fetch offset等于1
因为我已经本地LEO等于1了
我带参数的时候肯定是带1了
我又来请求你 请求你的时候
我们注意下这个图有些什么变化
high watermark=1
remote LEO=1 为什么
因为你在fetch offset的时候
我肯定是把remote的LEO=1
然后high watermark取这两个值
肯定是取两个人最小的
结果都等于一 我就更新它了
这个时候我再把response
再带回来 这个图应该错了
这个地方应该是带回来
leader HW应该等于1了
然后这个时候再把HW更新成1
我们详细看看它做了些什么事情
首先一样的
follower发了第二次fetch请求 leader会干啥
先读取底层log数据
发现啥也没有
然后更新remote LEO=1
为什么这次是1
刚刚说了fetch RPC带的参数等于1
我尝试更新
我肯定是把remote LEO就等于1了
尝试更新high watermark刚刚也说了
肯定是取两个最小结果=1
实际上把数据和当前分区的HW值已更新为1
注意这个图我这里有个bug
这里应该是1这写错了
已更新为1的数据发送给follower
follower干啥
收到fetch的response以后
它一样的先写入本地log
当然了啥也没写
因为没有东西
故follower的LEO还是等于1
然后更新follower的high watermark
这个时候发现本地LEO和的HW的取最小值都等于1
然后再把follower的HW=1
producer端发送消息后
broker端完整的流程
基本上我们就讲完了
此时消息其实已经成功地复制到
leader和follower log
并且大家分区HW都等于1
这表明了consumer能够消费offset=0的这个消息
因为high watermark
最终其实是对consumer可见的
也就是说你的leader
虽然LEO=10
但是你的HW=8
只有等于8的这条消息才能
被consumer消费到
我们再回到最早这个图
high watermark决定了
consumer能消费到哪
也就是说这个LEO虽然你可以先追加进去
但是你没有被成功更新到你的follower
这些数据对consumer是不可见的
因为它不安全
有可能是还没有处理成功的
这个是要跟consumer去联动的
我们再看第二种情况
就是producer和fetch请求交互的第二种情况
还有一种情况就是fetch请求的时候
follower是在我们刚才说的一个长轮询的里面
这个请求其实跟第一种情况基本上大差不差
它只不过说有一个唤醒的流程
简单来说还是一样的
leader写入本地log
同时更新leader的LEO
然后尝试唤醒那些暂存的consumer
然后把它这些fetch请求给响应了
响应完之后 其实就是跟之前一模一样
我就不展开讲了
然后也是一样要尝试更新分区的HW
我们了解到Kafka的复制机制之后
我们做两个总结
第一个就是何时更新LEO
我们分三个场景
第一个follower LEO什么时候更新
当follower副本端的LEO值
就是底层日志的LEO值
也就是说每当新写入一条消息后
且LEO值会被更新
也就是说我的follower的数据写成功了
我才去更新
什么时候能拿到这个数据
当我的follower发送fetch请求
去leader拖数据的时候
leader把这个数据返回给follower
follower就开始向底层写数据
自然而然它就更新LEO值 就是follower LEO
我们再看一下leader remote LEO
是什么时候更新的
leader remote LEO的更新
实际上在leader处理follower
fetch请求的时候来更新的
一旦leader收到follower fetch请求
比方说读到了 LEO等于1
它就会认为
先去尝试从自己的log中读取相应数据
但是在给follower返回数据之前
先去更新leader上的remote LEO
因为你的follower把LEO的版本号带过来
说明我知道你已经同步成功了
我就要把leader端的remote LEO给刷掉
更新成新值
leader的LEO何时更新
这个比较简单了
也就是说我leader只要写log成功了
我就把LEO值往前推进加1
所以这个是三个LEO什么时候更新
我们再看一下high watermark
分别什么时候更新的
先看follower
follower副本何时更新high watermark
首先 follower更新
high watermark发生在更新LEO之后
必须是先更新follower的
LEO数据同步过来的
然后我才会尝试更新high watermark
那也就是说follower向log写完数据
它会尝试更新它的HW值
但是第一次RPC肯定是更新不了的
第一次肯定是更新不了的
它的算法也是一样的
我在fetch response响应的时候
你会把日志写进去
你写完之后
你发现LEO和我响应的leader的HW值
取一个最小的去更新
也就是说数据同步
第一次RPC同步完之后
其实leader的HW其实还没有更新
但是我第二次RPC的时候
因为你本地的follower日志已经同步成功了
我自然而然leader的remote LEO更新
同时leader的high watermark也更新成功
这次response带回来HW的时候
它已经是最新值了
这个时候再跟你本地的high watermark值
进行一个比较
跟你的LEO进行比较
然后你就可以更新成1 这个是follower
leader什么时候更新HW值
leader在broker
其实它保存了一套follower的remote LEO
以及它自己的LEO
也就是说它在更新分区high watermark的时候
它会选出所有满足条件的副本
注意不是只有一个 如果有三副本就有三个
并且它们都在ISR里面的情况下
会比较它们所有人的LEO
包括leader自己的LEO
选一个最小的LEO
作为HW值
表示的是你的ISR
in sync的副本里面都同步了这条数据
我才能去更新high watermark值
这个时候producer向leader副本
写下来的时候会尝试更新
会更新leader的LEO
同时它也会去尝试更新一下HW
还一种情况下是leader收到follower
fetch请求的时候会尝试更新
更新逻辑也是一样的
就刚刚说的要满足所有的
ISR里面的LEO值
和remote LEO值 leader的LEO值
取最小更新HW
这个就是它的一个同步策略
这套机制它其实早期是有一些小缺陷的
我们看看它的数据的丢失的一些场景
在初始情况下
主副本写了两条消息 0和1
这个时候它假设它的HW=1
LEO=2 因为它LEO=2
是因为它写了两条消息
它有一个从副本B
它的LEOB=1 为什么
表示的是从副本0消息已经被确认同步成功了
假设属于这个情况
我们看一下这里的min.insync.replicas=1的情况下
这时从副本B向A发起一个新的请求
发了一个fetchOffset=1
就是主副本收到这个
请求之后尝试更新LEO=1
表示副本B已经收到消息0
尝试更新HW值
因为这个时候我们看一下LEO和LEOB
我们取最小等于1
也就是说不需要更新
将消息1及HW=1返回给从副本B
B收到响应之后写入日志并更新LEO等于2
把1给拖回来了
这个时候LEO就等于2了
再尝试更新HW的把它更新成1
虽然已经写入了两条消息
从副本写入两条消息
但是HW需要下一轮请求才会更新成2
注意这个high watermark
在从副本有什么用 其实用于截断
我要保证我的主从副本的数据是否一致
从副本的follower的
high watermark其实用于截断日志
注意这个第一轮RPC还没有更新
要第二轮才会更新
但是如果这个过程中从副本B挂掉了
重启之后从副本B会根据HW值
进行截断 为什么要截断
因为你只有用HW值
才能表示你主从副本是一致的
不截断这个数据有可能数据会重复
它要先进行截断
截断的HW处也就是把这个消息丢掉了
正常情况下你从副本重新去主副本A拖
又同步成功了 就肯定数据是不会丢的
因为你重新建立一个fetch offset=1
你又把1同步回来了
因为你要先截断 然后再同步
那肯定是没问题的
但是如果你这个时候从副本去主副本A拖的时候
主副本很不幸刚好挂掉
你这个时候从副本它就会升级为主副本
我们看一下这个时候主副本它的值
HW=1 LEO=1 LEOA=1
这个是它的主副本的一个状态
这个时候我们从副本
就从主副本变成从副本它重新复活
复活一样的 先要截断
截断完之后它去构建一个fetchOffset=1的一个请求
这个时候就尴尬了
这个时候把LEOA等于1了
然后返回一个消息
这个时候是不是这个消息1永远就丢失了
这个情况下其实数据就丢掉了
我们再看看一个数据不一致的场景
初始状态下主副本A已经写了两条消息
一开始写两条消息的时候
我们看到HW=1 LEO=2
LEOB=1 为什么等于1
因为还没复制过来
从副本B也同步了两条消息
它这个时候HW=1 LEO=2
这个时候从副本
向主副本发送一个fetchOffset=2的请求
主副本A收到这个请求之后尝试更新
HW=2把它更新成二了
并将该值返回给主副本
也就是在返回的过程中突然从副本宕机
也就是说这个时候HW=1还没来得及更新
这个时候你还没来得及更新它等于2
我们再假设主副本A这个时候也挂掉了
从副本B先变成主副本
就是它重新变成主副本的
你肯定是先截断 蓝色的这条就被干掉了
干掉完之后你主副本B
肯定要立马对外提供服务
这个时候假设外部写入的一个消息1
我们用不同的颜色表示
可能用黄色这个地方用黄色表示
等从副本A重新复活了以后
因为之前的HW=2
也就是说1这个消息并不会被截断
我去从fetchOffset=2去拖的时候
我的HW已经等于2了
我去fetch offset的
时候肯定是从2开始拖的
我拖到的是新的第三条message
拖的也就是这个消息
注意老的蓝色消息 这时候主从已经不一致了
这个时候其实就是主从不一致
因为你去拖的时候
我肯定是用我的LEO等于2去拖的
用我的LEO=2去拖的时候
我肯定是拖到这条消息
只不过说我们的offset你也等于2我也等于2
我肯定新的1不会拖过来
就尴尬了 因为我HW=2
也没有没有把老的消息截断
所以说又会导致数据不一致
基于这个问题
HW值核心它就是用于
衡量副本备份成功
以及是否会出现失败的时候
做日志的一个依据
然后其实也就导致了数据不一致的情况
在新版本Kafka它引入了一个叫
leader epoch的一个概念
epoch这个概念其实可以理解为版本号
也就是说每个集群有个集群版本号
我们把这个字段理解为那个版本号
这个offset表示的是什么呢
表示的是该版本号的主副本
写入的第一条消息的位置
举个例子
0.0表示的是第一个主副本从位移零开始写入消息
1.100表示的是第二个主副本
因为主副本从零开始
第二个主副本
也就是说这时候产生了一个leader切换
它版本号等于1了 从0开始的
第二个主副本的版本号为1
从offset100开始写消息
也就是说我切换为leader之后
我从哪开始写消息
主副本会将这个信息保存在缓存中
并定期写入到checkpoint的文件中
每次发生副本切换的时候
都会从缓存中查询这个消息
加入这个机制以后
我们再来看一下它数据丢失的一个场景
我们引入这个版本号之后
我们先看一下第一种场景
主副本A和从副本B还是一样的
跟之前那个场景一样
我们看一下这个地方多了一个
版本号和offset
这个地方也多了个版本号和offset
假设这个时候从副本B重启之后
以前是这样的
新版本啊它重启之后
从副本B向主副本A发送同步请求
但是它接口名字改了
叫OffsetsForLeaderEpochRequest
这个时候主从副本相等
注意这个版本号没有产生leader切换
所以它们的版本号一致
如果版本号一致
从副本B实际上它不会做任何的截断
我们两个版本号完全一致
你返回给我lastOffset=2返回这个值
然后我就不需要截断了
所以数据是不会丢的
这个其实是非常有意思的一点
以前是一重启先截断
现在我先不截断
我先发一个请求问一下你
最新的LEO等于几
因为我这个时候截断的依据
根据你LastOffset=2的值
截断比它大的
结果发现没有比它大的我就不截断
这个时候就恢复了
这是第一种情况
我们看第二种情况
从副本B向A发送fetchOffset=2的请求
发送的时候 A宕机了
A宕机以后从副本B成为主副本
并更新epoch值
这个时候把它更新成等于1
把epoch=1 把offset=2
注意还是不截断
我们看不截断的情况下会不会出问题
不截断以后A这时候恢复
主副本A就变成从副本A了
我还是一样地发送一个
fetchOffset=2请求 发送的时候
因为我的从副本变成主副本之后
我并不截断
我并不截断之后
我就强行把我的HW=2
把它强行更新成2 好
这个时候你调我这个请求的时候
我就返回给你HW=2
返回给HW=2的以后
这个从副本
就会把HW=1更新成HW=2
所以你看大家都是01
它这边也是01 颜色也一样
是不是没有任何丢失
所以说它的机制就在同步的时候引入接口
或者说强制更新high watermark
导致数据不会出现问题
这个是非常有意思的
但是注意从副本B起来之后
向主本A发送OffsetsForLeaderEpochRequest请求
如果这个失败
如果主副本A这个时候同时也宕机
这个消息1还是会丢失
也就是说会把它截断掉
我们看一下数据不一致的场景里面
我们再来看一下
假设从副本B
恢复之后向A发送了
OffsetsForLeaderEpochRequest请求
发送请求因为主副本刚好也宕机了
因此从副本B就变成主副本
并把消息1给截断掉了
此时假设我又接受了一个新消息的写入
就变成黄色的 注意
因为版本切换之后我们看一下
epoch=1 offset=1
蓝色1被截断了
所以这个offset等于1
副本A这个时候它又恢复了
它就从主副本变成了从副本
它变成从副本以后
它就会发送OffsetsForLeaderEpochRequest请求
请求的值小于主副本B的epoch
也就是说你起来的时候你以为你是leader
结果你的版本号等于0 我这等于1
这个时候我就把最新的lastOffset等于1返还给你
我的上一次成为leader前
我的最后写的那条消息offset等于几
等于1 我就把这个返回给你
返给你之后你就要把蓝色1给截断掉
这时候我再重新发起fetchOffset=1请求
过程中还会更新epoch的值和offset=1
引入epoch之后避免的数据不一致
我们之前那个做法其实是会不一致的
但是目前能解决数据不一致的问题
我们刚刚讲的还是会存在数据丢失的一个场景
我们一定要稍微注意一下
我们再来看一下Kafka的数据可靠性
其实也是蛮有意思的一个点
对于某些不重要的数据可靠性要求不是很高
能容忍的少量数据丢失
有些场景比方说
我们做日志传输要求不是那么高的情况下
你没必要等ISR所有follower全部成功
那只有被ISR中所有replica同步消息才会commit
但producer发布数据的时候
leader并不需要ISR所有的replica
同步数据才确认收到
也就是说我的生产者在生产消息的时候
不需要等它们全部成功我才返回
我可能发到leader了我就想返回
我这样吞吐是最高的
这个机制也是Kafka抛给我们调用者的一个选项
我们看一下几个选项
这个叫producer required.acks
等于0的情况下producer不等待broker的ACK
这个提供了最低延迟
就是说我无脑发给你就返回了
也不需要返回就发给你就行了
Broker一旦收到数据还没写入磁盘就已经返回
当然了就是broker故障的时候可能会丢失数据
因为你可能还没刷盘
第二种选项等于1
producer等待broker的ACK
要等partition的leader落盘后才会返回ACK
如果在follower同步成功之前
leader故障 会丢失数据
因为你只有一个副本
还有一个叫all等于-1
它是producer等待broker的ACK
但是partition的leader和follower
全部落盘成功后才会返回ACK
就是大家全部都刷盘成功
但是在broker发送ACK时
leader发生故障
可能会导致数据重复
也就是说你写给leader成功了
然后follower还在同步
但这个时候如果leader发生故障
你可能会
选一个follower重新作为leader
有可能同步成功 有可能没同步成功
但是leader因为故障了
你的producer的client会报错
你这个时候其实是不知道到底写没写成功
如果你再写一次数据可能会重复
所以这个幂等一直我在强调还是要解决
就像我们现在MySQL
你触发一个MySQL事务
它虽然能保证原子性
但是这个事务到底是成功还是失败
你是不能100%保证的
你明白什么意思
就是我们提交一个MySQL事务
如果这时候你网络中断
你其实并不知道事务到底是调commit成功以后
但是commit返回给你的时候
网络出现了问题
你其实并不知道这个事务到底是成功
还是提交还是失败
所以业务幂等一定要注意
一定要你自己处理
我们再看一下
如果我们要提高数据的可靠性
在设置request.required.acks=-1就是ack all
还有一个值叫min.insync.replicas
默认情况下等于1
就只需要一个副本
写成功了就返回
这个其实可以在broker
或者topic层面进行配合设置
这样才能发挥最大的一个功效
min.insync.replicas这个参数表示的是
设定ISR中最小的副本数是多少
默认是1
而且仅当required=-1的时候
这个参数才生效
也就是你要设置ack all才会生效
如果ISR副本数少于
配的min.insync.replicas的数量的时候
客户端可能会抛异常
就是NotEnoughReplicasException
就是说如果你有大量的follower数据delay
你可能就是写数据是写不成功的
其实就跟Raft一样
如果你有过半数副本都失败是选不出leader的
所以这样的话你两个参数一配合
才能保证一个最高的数据的可靠性
就是说我们需要ack all
同时我们要min.insync.replicas至少大于1
这两个参数是非常关键的
我们再来看一下
我们再举个例子
为什么在required.acks等于1的情况下
会产生数据的一个丢失
producer把数据发送给leader
leader的写日志成功也是把4和5写成功了
这个时候返回客户端成功
因为你acks=1写到leader成功就返回
ISR的副本还没来得及获取消息
如果这个时候leader故障
这个时候消息其实就丢掉了
我们看一下第一步
producer发送消息45
发过去 第二步
leader本地写日志成功
同时客户端返回成功
第三步follower准备到leader中fetch消息
就还没来得及fetch的时候
第四步leader就失败了
这个时候这两步操作
follower获取失败 并进行了选举
这时候选leader时候可能follower
把原来的follower 1变成一个新的leader
这个时候LEO等于3
producer这个时候
发送的4和5其实就丢失掉了
因为你的acks等于1
你producer就直接返回了
我们再看一下
在required.acks=-1的时候
并且我们的Kafka要默认等于同步的发送模型下
且复制因子就是replication.factor>=2
且min.insync.replicas>=2的情况下
这几个条件全部满足的情况下
才彻底不会丢失数据
有两种典型的case
在acks等于-1的情况下 数据发送到leader
ISR的follower全部数据同步后
leader这时候挂
这个时候选出新leader其实数据是不会丢的
因为数据还是存在
我们再看一下它为什么会ok
第一步 producer发送消息到leader
4和5发过去
第二步leader写本地
第三步follower准备到leader中去fetch新的消息
第四步 follower全部同步完成
同时更新HW值
就更新high watermark
返回客户端成功的一个标志
这个时候leader如果发生crash
从ISR中选一个follower作为leader
这个时候当follower 1成为leader的时候
数据是没有任何丢失的
所以这个其实是非常有意思的一个点
我们再来看一个更有意思的一个点
在ack all的情况下
数据发送到leader 部分ISR副本同步
leader这个时候此时挂掉会产生什么样的行为
我们看一下
第一步我们的producer发送消息
到leader还是一样的
第二步写日志写成功45
第三步follower准备到leader中fetch消息
第四步注意follower 1同步完成
但是leader发生crash
返回客户端错误
因为leader错误了肯定会报错
这个时候follower 2尚未同步成功
因为没有完全同步
因为它还没有完全同步
所以说你producer失败了你肯定会重试
你可能会重新发
但是这个时候follower1
它成为了一个新的leader
就是我们的所谓的第五步
成为新的leader
我们第六步就是新的follower
准备到leader上去拖消息
也就是它会去拖消息 拖过来
因为你的producer收到的是失败
所以它会继续重试
这个时候
如果它再发45过来
是不是4和5
就冗余了 double了两份
所以这个acks等于-1的数据重复的一个问题
我们再来讨论一下
关于high watermark的进一步讨论
还是考虑像上图
acks等于-1部分ISR副本同步的另外一种情况
如果leader在挂掉的时候
follower 1同步了消息45
follower 2同步了消息4
同时follower 2被选为leader
注意它不是选最多的数据成为新的leader
它不一定的
这个时候follower 1多出来的消息5如何处理
就是选leader的不一定是选数据最多的
我刚刚说的我们刚刚那种情况下
就是follower 1同步的45
follower 2只同步了4
如果这时候你把follower 2选为leader
是不是你之前follower 1的消息5就是多出来的
这个是不是多出来的
之前follower 1的5是多出来的
这个时候其实我们之前讲了
high watermark就有用处了
一个partition中的ISR列表中leader一个
high watermark表示的是
ISR列表中副本中最小的LEO
类似于木桶效应
水位取决于最低那一段
ISR个副本的LEO不一致的时候
其实此时leader的挂掉选举新的
leader的并不是按照高低选举的
而是按ISR的顺序选举
所以它就会产生这种情况
因为我们上一种情况下
因为你如果选它作为leader的话
这个5会不会不一致
不会 它会把它斩断掉
但是你这个时候如果producer继续发送
重试发45 其实还是会数据产生重复
数据是没有丢的
但是它会产生数据重复
这个是非常有意思的一个点
一定注意它不是选最多的
我们再看看为什么Kafka性能高
我们其实简单来说几个层面
首先 partition级别是并行的
从架构层面来讲
partition级别它就是一个并行的
因为broker是本地挂盘的
所以broker级别并行是第一个级别的并行
第二个级别是本地
broker可以挂好多个硬盘
所以它硬盘级别可以并行
consumer端其实是
不同的partition它支持rebalance
所以这个效果还挺不错的
第二个就是ISR这种机制
虽然说会存在数据丢失的一个风险
但是它不会出现数据不一致
因为通过引入了leader的版本号的策略
但是ISR它有个缺陷
它还是会出现数据丢失
它没有Raft那么好
并不是像Raft这样过半数副本成功
但是它的确ISR这套机制
会使得我们使用的机器数量要少得多
从IO层面Kafka做了哪些优化
比方说batch的读写
批量读写 你去pull的时候
你不是只拿一条消息
你可以一批拿100个
如果你的生产速度一下灌了100条进去
我pull的间隙我就可以一次batch的去读写
另外就是利用了磁盘的顺序IO
buffer IO page cache
在Kafka的实现还使用了zero copy
我就不需要把数据从磁盘
读到内存再发送网络 我可以直接磁盘
sendfile到我们的网络端
另外就是压缩
另外就是我们的Kafka支持
在producer端去压缩的
数据可以最小化
等等一系列原因
这就是Kafka高性能的本质
看一个分布式的架构其实整体还是非常复杂的
我们课后的reference其实是非常多的
我建议大家还是要认真阅读
好今天的内容 咱们就到这了
再见