我们今天开始第12周的内容
这周主要是介绍一下我们
微服务里面经常用到的消息队列
我们前面介绍的主要是RPC
RPC大部分情况下都是所谓的同步通讯
那么Kafka我理解为它是一种异步通讯的机制
Kafka还有很多的优势
然后我们这一节主要把Kafka的一些优势
包括我们前面在介绍评论系统 播放历史
其实也是结合Kafka的一个形态
它的一些机制和原理
来进行一些系统的架构上的优化
这一章我们会更深入到它的底层的一些原理
它的存储实现
它的可用性的一些设计
它的数据的副本的机制
那这些其实也是非常重要
我觉得通过我们对Kafka的一些理解
那也是帮助我们去
真正的去思考一个
大型的分布式系统是如何去设计和实现的
大概会分为几块
第一块就是Kafka一些基础的概念
因为不是所有的同学都在使用Kafka做消息队列
我知道可能有些同学使用Redis做消息队列
那么大家知道用Redis做消息队列它有些什么坏处吗
第一 它的的吞吐是受限的
因为Redis双线程的模型
在这种海量的日志传输 行为传输
还有它的一个带宽
就是带宽很密集 字节很大的情况下
Redis吞吐会下降的非常厉害
所以它不太适合特别大体量的
尤其是这种大吞吐的场景
第二 Redis本身
因为它是没有很优秀的一个持久化的方案的
它是一个cache
它是一个易失系统
Kafka其实它是带存储的
存储指的是什么
我们一般像MySQL
像KV
像Kafka我们都会把它归类为叫storage
因为它是带存储的
会落地到磁盘会持久化的
Redis它是易失的
我们虽然知道Redis
是有类似像RDB
AOF这种类似像快照
AOF的可能是每来一个指令就会
append到它那个文件里面
但我觉得都是某种意义上是有些缺陷的
它无法被很好的scale out出去
所以我在生产环境是不建议大家
用Redis做消息队列
因为你的整个Redis
如果内存crash掉
以及整个进程crash掉
那么你的整个恢复的时间也会非常长
那么也会
使你的系统变得非常脆弱
其实我会分几块
就是Topic和Partition
是Kafka最核心的两个概念
消息队列里面其实一个业务就是一个主题
主题是由什么构成的呢
它实际上是由分区构成的
接下来就是像Producer
Consumer 消费者 生产者
然后以及在Kafka内部
所谓的Leader和Follower的机制
以及我们还会讨论一下Kafka的可靠性
因为我们要用它做消息队列
万一数据丢了怎么办
所以我会告诉大家它不同的一些参数组合
如何去完成它的一个数据的可靠的传输
最后就是关于Kafka的一些性能优化
或者说我会在这一块也会讲一讲
我们对Kafka做了一些优化或者改动
或者说一些展望
这些可能都是我会去跟大家去聊的
我们先看一下Kafka的一个最基本的概念
首先什么是Kafka
那么Kafka
它实际上目前是被多家不同类型的公司
作为多种类型的数据通道和消息系统使用
就是一个是做消息通道
做了一个数据通道
比方说做我们大数据生态里面的这种日志流
比方你可能要采集日志
ELK里面可能就会使用Kafka
做一个程序的那种debug日志或者是
info日志的一个cache一个缓冲
一个存储一个暂存
第二块就是说像我们的用户行为的一些埋点
我们可能要做大数据的分析
埋点
然后像event tracking一些事件的追踪
可能也会使用Kafka作为通道
还有就是说像我们的消息系统
消息系统可能是业务上的一个生产的一些
消息队列这种消息系统来使用
像我了解到大部分公司基本上
在国内基本上主流的互联网公司
它的行为流数据基本上都是使用Kafka来做的
我们一般来说对网站
或者对你的APP在报表中做一些用户行为分析
用户行为分析指什么
比方说我要选两个页面
我要看这个页面的一些曝光
点击 PV
这种很简单的页面级别的一些行为分析
我可能会看一下
两个页面之间的它的一个转化率
点击的转化率
甚至我会看一下几个页面的访问的路径它的漏斗
我们叫路径或者叫漏斗
这个其实也是在数据分析里面很常见的
这些都是会使用Kafka来作为消息的一个传输通道
行为的话一般来说会包含像页面的
一些访问量的PV 页面的曝光
页面的点击这些行为事件
在实时计算中其实也会使用Kafka
因为我们现在无论是用Apache的Spark
Storm或者Flink
你想你要做实时计算
那么你肯定是要消费数据源
通常大家第一优先都是支持Kafka的
所谓的一个Data Source
第三块其实是在业务的消息系统里面
比方说通过一个这种发布订阅的消息解耦
多组微服务
我们举个最典型的例子
比方像我们账号系统登录成功以后
每天首次登录会发放积分
账号跟积分实际上两个微服务
它不是那种强绑定关系的一个业务形态 对吧
对于账号系统来说
积分可能只是它的一个旁路
或者说对于积分服务来说
账号只是它的一个载体
所以它两个系统既然不是强相关
能不能不让两个微服务直接产生依赖
而是让它们之间的数据产生依赖
什么叫数据产生依赖
也就是说你要约定一种数据格式
一种数据的schema
然后通过发布一个消息
比方说我首次登录成功了
我发布一个消息给它
我的下游消费系统
我的下游的积分服务
消费这个消息进行一个积分的增加
其实这两个系统其实完全解耦的
那么还有一些场景主要是消除峰值
我们之前在介绍评论系统也好
还是播放历史也好
我们都多次强调过
因为在我们海量的一个流量的请求
或者说写入的情况下
我下游最底层的存储它的scale out的能力
相对来说是偏弱的
我的MySQL也好 我的TiDB也好
你要去实时scale out
这个肯定是很难的
不像我们发布一个应用
你可以立马开一个节点
然后你的流量就会分摊了
你的存储其实是很难的
其实是很困难的 怎么办
我们其实核心就是要解决
流入的速度和持久化落盘的
落到我们的存储上的两个速度的差速
一个快一个慢
就流入的多消费的慢
怎么去消除峰值
实际上就是要使用我们消息队列
做一个中间的暂存 做一个缓冲
缓冲 我就是产生一些消息堆积
那么一旦有了消息堆积
是不是这些流量被平滑到你的下游的消费系统
它的一个最大吞吐能力啊
我的下游最大吞吐是多少
我的消费速度就是多少
这样的话缓冲可能会越积越多
最终肯定要全部消费完
这也是很常见的一种使用模式
Kafka最早是由在LinkedIn
它开发的并开源的一个分布式的消息系统
因为它的分布式及高吞吐率被广泛使用
所以现在也与像Hadoop
刚刚说的Spark Storm
Fink去深度的去做了很多集成
这个是Kafka的一些背景知识
Kafka它本质上就是一种分布式的
基于发布订阅的
基于主题的这么一个消息系统
它有如下的几个设计目标
第一
它是以时间复杂度为O(1)
方式提供消息持久化能力的
那么即使对TB级以上的数据
其实它跟数据规模已经没有关系了
它的整个写入速度始终都是O(1)了
无论你多大数据量
那么它始终能够保证
常数的时间复杂度的访问性能
因为它是顺序写
然后顺序消费
所以它始终可以保持一个常量的访问性能
这其实是非常优秀的一个特性
我建议大家去看一下一本书叫做《DDIA》
中文名《数据密集型应用系统设计》
这本书介绍了大量的各种的一些数据结构
存储上的一些数据结构
比方说LSM Tree
我们知道MySQL使用B Tree
那么B Tree和LSM Tree对比的话
其实也会涉及到我们的顺序IO的一个问题
那么你了解到顺序lO的话
你就要了解一下磁盘的它的一些原理和特性
这个大家可能要后面去发散性地一些阅读
系统性的一些积累
这是第一个
它怎么去实现这样的一个时间复杂度
我后面也会去介绍
第二个就是它的高吞吐率
即便在我们的这种X86非常廉价的商用机器上
也可以做到单机每秒上10万的消息传输
还有它支持
Kafka Server间我们也叫Broker
就是Kafka Server间的一个消息分区
以分布式的方式去消费
并且保证Partition级别内的消费的顺序
举个例子 我们看下这个地方
一个Producer发布一个消息到一个Topic
它有好多个消息123456
注意这123456
可能是被分散在两个Partition内
假设我有两个Partition
那么可能123在第一个Partition
456在第二个Partition
那么123456
全局它的的确确很难保证消费顺序
但是每一个Partition内它都是顺序的
123在Partition 1
456在Partition 2
我又可以开好多个消费者Consumer
去消费这些消息
第四点就是它能很好的可以支持这种离线处理
实时处理 我实时消费它
这个其实都是很方便的
另外它有还不错的一个scale out的能力
比方说我可以很方便去扩展我的Kafka的Server
去扩展我整个集群
因为在我们的生产规模里面
随着Kafka集群规模越来越大
当然也会遇到很多的
像快手也好 像我们也好 像美团也好
都有类似Kafka在大规模集群下可能遇到的一些问题
大家也可以去搜一搜关于滴滴 美团 快手
都会有搜到一些类似的文章和挑战
回到消息队列
我们介绍了很多它的一些基础知识
我为什么要去使用消息系统
有没有认真去思考过这个问题
这个问题真的值得我们去认真想一想的
我觉得有这几个方面
第一就是解耦
消息系统
在处理过程中
它插入了一个隐含的基于数据的接口层
我们之前讲的微服务它是什么
两个人依赖是什么
它是基于API的依赖
现在你通过消息队列
虽然说叫服务的API依赖没有了
就是两个微服务直接依赖没有了
但是它其实是一种间接的隐含的依赖关系
那么我们把这种依赖关系叫做什么
基于数据的接口的依赖
也就说你们俩必须认准同一个数据的Schema
比方你用Protobuf定义了一个Struct类型
定义了一个message
它有哪些字段
那么其实你们大家都是
隐含的在数据层面是有一些依赖的
那么两边的处理过程
其实都要实现这个数据接口
也就说你要认准数据的Schema
这样的话好处是什么
它允许你独立的去扩展
或者是修改两边的一个处理过程
比方说你的生产者和你的消费者
完全可以是两个微服务 两个进程
你可以分别去改
你爱怎么改怎么改
你只要确保它们遵守同样的一个接口约束
那么基于这样的一个消息发布订阅的机制
可以联动多个业务子系统
就像我们刚刚说的账号和积分的关系
就是我的登录的这些消息这些主题
可能还会涉及到我的安全反作弊等系统
其他的一些下游子系统
都去消费这些登录的一些记录
所以它是很容易被多个下游子系统去消费的
能够在不侵入我们主干
账号的业务逻辑情况下
分步编排和开发
同时也是要保证账号登录的记录
让下游多个子系统都要保证数据一致性
当然它也是最终一致
也就是说对于登录这个消息很多人感兴趣
大家都会去订阅并且去消费它
那么其实间接的就完成了
一个数据一致性的一个工作
那只不过说它是一个最终一致
因为我们经过消息系统
异步的去消费这个逻辑
第二个 我们叫冗余
有一些情况下
我们知道同步接口
你调它失败了
你怎么办
你只能重试对吧
如果接口重试得过于频繁
或者说重试的次数过多
失败的耗时其实也是会累加的
那么这个时候
你的下游的用户或者你下游消费的Client
可能是无法接受的
那么应该让我们的处理过程
是不是经过消息队列之后
如果同步接口失败
是不是很难处理
如果你是在一个异步系统里面
那么在失败的时候 它是不是很容易
去对这种失败的数据进行一些处理
所以说有些情况下处理数据的过程会失败
除非你把这个数据持久化了到某一个地方
否则你在同步接口里面
你不重试就会丢失
甚至你重试如果你的进程crash掉
那也会丢失
那消息队列的把这些数据进行持久化
直到它们已经被完全处理
通过这样的一个模式
这样的一种方式规避了数据丢失的风险
同步是很难处理好的
那么许多消息队列
通常采用的就是叫插入 获取 ACK
ACK就是删除的方式
在把一个消息从消息队列删除之前
你要确保它你的业务的下游的子系统
能够完完全全消费掉
经过你的处理系统可以处理完毕
从而确保你的数据可以被安全地保存
直到你使用完毕
也就是说你像RabbitMQ发送一个
Message到Queue里面
你可以通过ACK去确认消费完了
这个是冗余
那么第三个是扩展性
因为消息队列解耦了你的处理过程
所以增大消息的入队和处理的频率是很容易的
我可以扩展Kafka的集群规模
或者它的Scale out能力还不错
那么只要增加消费的处理的过程即可
所以它的扩展性非常好
不需要改代码
不需要调参数
也就是说如果我的消费有瓶颈了
我只要快速的去扩展我的Consumer
如果我的Producer有瓶颈
我是不是可以扩展我的生产者
或者如果说我的消息队列有瓶颈了
我只要扩容我的消息队列就可以了
所以它整个扩展性也是非常好的
那么下一个点就叫灵活性和峰值处理能力
在访问量剧增的情况下
应用仍然要继续的保证可服务
保证稳定
但是这样的突发流量并不常见
我们为此要为这种突发的流量投入的资源
等待这些峰值流量肯定是无比浪费的
也就是说你每天就有一个
十几 二十分钟或者每天就这么一个小时
会迎来一大波的这种请求流量
如果你剩下的时间
你的机器都要为了准备最大的峰值流量
去考虑它的容量规划的话
是不是就产生一个巨大浪费
是不是有可能使用消息队列
消除掉这些峰值
让我的存储不需要扩容的情况下
仍然满足我的业务的这种大量的一个写入
当然了其实也是跟你的业务形态有关
如果你的业务形态不接受这种异步消费的模式
你可能只能去扩容了
我的观点就是说通过引入消息队列
其实是某种意义上让我们的架构更加健壮
它的鲁棒性会好得多
那么在使用消息队列能够
使关键的组件顶住突发的访问压力
只要Kafka能扛得住
我下游慢慢消费就好了
不会因为突发的这种超负荷的请求
导致我的下游数据库被打爆
这个是非常关键的一点
当然了我的理解里面
峰值的一个处理能力
还有几个解决手段
希望大家也知道
峰值一般我们叫削峰
什么变得多 我就让它处理的更平滑
这是第一个
第二种思路是什么呢
叫自动扩缩容 Auto Scaling
根据我负载的一个情况
自动去扩展我的无状态的应用
当然我的Kafka扩容可能会相对困难一点
可能我要去扩Broker
因为可能是CPU密集
如果是IO密集可能要扩IO
这个相对会难一点
这个是另外一种做法 就是叫Auto Scaling
让我的处理节点变多
第三种大家其实往往没想到
叫降级或者叫限流
因为有时候比方我们在做双11
我们可能为双11付出的资源代价是非常高的
那么如果这些资源是常备的
hold在那里非常浪费
其实我可能只要扛住
双11整点的那么十几二十分钟的这些流量
让它慢慢排队甚至降级
让用户待会重试
说白了购物车的东西最终还是要被清空的
还是会买的
那也是一种有损服务
但这样的话节省的机器资源是少很多
当然我觉得可能还有一种思路
比方说在阿里双11
它可能还会有大量的公有云的机器和资源
这些资源其实也可以产生一个混合调度
我们叫混合部署
就是在线离线业务混部
不同的业务形态混部去蹭资源
最终把我的平均利用率做高
这也是一种思路
这样的话我的机器
也可以在那个时刻被大规模使用起来
我觉得这也是一种
还有叫可恢复性
可恢复性是什么
系统的一部分组件失效的时候
不会影响到整个系统
消息队列它其实间接的降低了进程间的直接耦合
这两个微服务直接耦合RPC
你要做很多事情
如果它失败了 你要做融断限流降级
重试等等
但是如果你依赖一个消息队列
只要你的消息队列足够可靠
消息队列也可能会挂
但是我们正常情况下
消息队列因为它做的事情非常简单
它就是个纯的一个中间件基础设施
它不带业务逻辑
当然它也要做很多自己的一些
高可用的一些设计
那么在这种场景下
如果你是使用消息队列的话
如果你的消费进程挂掉了
其实对我的主流程业务没有任何影响
因为它是个旁路 一个异步的旁路
所以我认为加入队列中的这种消息
它可以被持久化到哪里
Kafka中 也就是说数据不会丢
那我的消费者其实只要快速的重启恢复就好了
所以它可恢复性会变得不错
另外就是一个叫顺序保证
因为我们知道用消息队列
其实很多场景可能都需要
数据的一个顺序消费
大部分情况还都是很重要的
因为大部分消息本来也是有排序的
有没有办法能够可以顺序消费
我们可以想象一下
如果你自己你去设计一个消息队列系统
如果你的吞吐流量非常大
每一条消息绝对顺序
你觉得你的消费的并行速度能起得来吗
就说你第一条消息没处理完
你就不能处理第二条
你整个吞吐肯定是上不去的
Kafka我觉得很接地气
它设计的一种叫做我们Kafka的一个Topic
那么它下面是有很多个Partition来组成的
Partition可以理解为它就是一个文件
你选一个Partition去写这个数据的时候
肯定是追加到尾部
一个Partition的消息肯定是有序的
所以它叫什么
全局无序 局部有序
我觉得这也是非常接地气
因为我们很多场景里面其实是按照
某一个维度让它顺序消费可能就足够了
最后一块是缓冲
缓冲 在任何重要的系统中
都会有需要不同的处理时间的
一些任务一些元素
消息队列通过一个缓冲层
来帮助任务的最高效率的执行
写入队列的这些处理会尽可能的快速
缓冲有助于控制和优化
数据流经过系统的速度
这套话其实听起来有点懵
其实简单来讲就是说
你的任何的数据都会进入到一个缓冲
那么Kafka其实就是做什么的
就是做缓冲的
就像我们大数据做日志传输里面
我们为什么使用Kafka
其实也是第一是为了削峰
解决我们落地的HDFS
和我们的日志流的两个吞吐不匹配
那么Kafka在这个里面作为一个什么角色
它削峰 其实它也是缓冲
因为它有持久化
这个缓冲层
让我的任务执行的效率其实是最高的
因为它是在满足你下游子系统
最大吞吐的情况下
去捞取去消费消息的
当然我还可以支持多路缓冲
这也是Kafka的一个特性
也叫多Group去消费
最后就是异步通讯了
很多时候我们不需要立即去处理这个消息
可能是一个延迟的消息
我使用消息队列自己在做一些小小的改造
加上一些排序的结构体
其实你可以想象一下
我们Kafka其实是不支持所谓的
Delay的message
就是延迟消息消费 或者说定时消息消费
基于Kafka其实是很容易实现这个系统的
你可以自己想一想
其实可以抛一个问题给你
因为这些消息你不需要立即处理
那么你放在消息这些里面可以暂存
之后再把它取出来消费其实是完全可以的
所以这个叫异步通讯的一套逻辑
了解完Kafka的一些基本概念之后
我们回到它最核心的两个概念
Topic和Partition
Topic 主题经常会被问到一些问题
到底我是多个微服务用一个Topic还是怎么着
其实我们也很容易想到
不同的微服务
肯定是有不同的Topic
因为我们一定是要去中心化
包括元数据去中心化
元数据就是像Topic这种东西
像微服用自己独占的cache DB
就是一种独占消费
同样的道理
Topic我们可以理解为是一张表
可能微服务要用不同的Kafka集群
可能只需要用不同的Topic就足够了
因为Kafka吞吐非常好
就是说搭一套集群
可以里面创建好多个Topic
然后去给业务的使用
所以我们原则上
一个微服务至少是一个Topic
那一个业务域里面
比方说一个微服务它可能有不同种类的消息
对于我来说
我倾向是建不同的Topic
除非是什么一种特殊的情况
它是一个完整闭环的一个业务流
那么在这个业务流里面有多种状态
多种状态可能需要串行消费
这个状态可能还有一个上下游的依赖关系
这个时候没办法
必须投递到一个topic
那么也就说这个topic里面
可能存在几种不同类型的Schema
这个其实是很头疼的一个事情
就是说你这个topic里面
实际上有三四种不同的JSON
可能你是用Protobuf
可能是三四种不同的Protobuf
那么如何可靠地识别
这个topic里面为什么有这么几种不同的
JSON Protobuf
其实是很困难的一件事情
是一种很困难的事情
但也不是说彻底没有办法
其实也有一些办法的
比方说你可以把你的原始消息
在外面再包一个特殊的元数据
比方说你告诉它
它是类型一类型二类型三
你可以前面加一些字节表示它是
头两个字节可能表示的是不同的消息类型
这也是一种做法
简单理解就是说
最终这个主题你需要
把这类消息
放到哪一个消息队列放在一个queue里面
我们把一类消息按照主题来分类
其实有点类似于数据库的一个表
为了使得Kafka的吞吐率可以线性提高
那么物理上我们把Kafka分成
一个或者多个Partition
对应到系统上的
就是对应我们Linux操作系统上的
它其实就是变成了一个或者是若干个目录
也就是说我这一个
这个Partition可能是一个
目录或者一个文件
Partition 1是一个目录
Partition 2是个目录
不同目录里面肯定是有不同的文件的
所以看一下
这些消息它都是有顺序的
所以它局部有序
但是全局Partition 01234
它消息能还原出之前的顺序吗
这个是比较困难的
所以不同的主题就是不同的表
这个就是主题分区的一个关系
我们除了了解topic
其实topic跟broker强相关
我们再来看一下broker是什么
Broker可以理解为就是Kafka的一个server
Kafka集群包含一个或者多个server
有一组nginx 一组Kafka
一组broker 一组server
那么每一个服务器节点被称为一个broker
我们看一下它的关系
大家注意看Broker 1里面它实际上有
Topic1和Topic2
Broker 2里面也有Topic1和Topic2
里面可能有Patition1和Patition2
所以在这里面是什么意思
broker负责存储topic的数据
举个例子 如果某一个topic
有n个Partition
集群有n个broker
我们可以想象一下
你的topic有n个partition
假设有三个partition
你的集群有三个broker
我们是不是说
可以每个broker存储该topic上一个partition
刚刚好
可以均分完毕
那么这个就是broker的意义
计算和存储是不分离的
那么也就是说一个broker
对应一个partition
对于某一个topic下的某一个partition
我刚好有三个broker
我又有三个partition
是不是刚好一人拖一个
是不是我又把它分在不同的机器上
它的吞吐能力也会变得更好
那么所以从scale out的角度的性能考虑
通过broker Kafka的server的更多节点
带来更多的一个存储的通道
更多的存储的IOPS
建立更多的partition
把这些IO能够尽可能负载到更多的
Kafka的broker的物理节点上
提高它的总吞吐的IOPS
散在不同的物理机上
因为每个机器上的磁盘是有限的
那么你可以散在不同的物理机上
总体的IOPS变强了
也就是说topic的写入能力
会变得非常的强
这个是第一
第二
从scale up的角度考虑
一个节点拥有越多的物理磁盘
就是我可能一台机器挂16块盘
我可能挂32块盘
我挂非常多的盘
它也是一种扩展能力
它可以负载更多的partition
也就是说单机可以放越多的硬盘
那么就可以放越多的partition
分布式我有越多的Kafka的server
更多的Kafka的broker节点
我又可以什么
有更多的IO吞吐能力和计算的能力
因为Kafka整个消息还需要耗费一些CPU
我们再来看一下这个场景里面
它到底还topic怎么分配的
如果我的topic有n个partition
集群有n+m个broker
那么其中有n个broker存储
该topic的一个partition
因为broker数量是大于partition数量的
所以有n个broker要存储就刚好均分
那么剩下的m个broker
不存储该topic的partition数据
因为不够分了 一人一个
我最小粒度的分配就是一个partition
那么第二种情况
如果某个topic有n个partition
集群中的broker数量小于n
也就是说我的Kafka的server节点数量
是小于partition的
那是不是意味着
有那么一个broker可能存储该topic的
数量比别人多一些
别人都存一个 你可能得两个
因为你分完之后还有多
多完之后你可能要把多出来放到
剩下的broker里面去
所以topic只是一个逻辑概念
真正在broker间分布式存储的是什么
是partition
每一条消息被发送到broker中
会根据partition的规则
这个是我们Kafka SDK里面非常重要的一个点
你写入到Kafka的message的key是什么
按这个key去可以做一些sharding
或者说做一些负载均衡的策略
这个就是partition的规则
你的选举规则决定了你的
带这个key的message
最终会存到哪个partition中
那么如果你的party的规则设置的非常合理
最终是非常均匀的
我们做个实验
比方说三个broker和一个topic
没有副本 没有replication
然后它是异步模式
然后这个时候有三个producer
三个Kafka的节点组成了一个集群
那么我现在有三个producer
产生100字节的消息去进行一个压测
然后我们压测的时候发现
单partition的数量小于broker的时候
我有10个broker
我只有3个partition
那么这个时候我再去提升partition的数量
它的吞吐会越高
性能是呈线性增加的
注意你看这个地方
随着partition的数量增加
它的吞吐会越来越好 为什么
有可能分担的broker
以前三个partition就是3个broker
以前8个partition就8个broker
那么8个broker
它的总体的IOPS和总体吞吐能力要强的多
所以它可以线性上升
好 我们再换一个场景
Kafka将所有的partition
因为会均匀地分布到broker上
你只有两个partition的时候
当然只有两个为它服务
充钱就变强了嘛
我换另外一个case
当partition的数量多于broker数量的时候
就是你这时候broker是更少的
总吞吐量并没有提升
就是说我随着我的partition数量一直
在持续增加 比方说10个broker
我现在partition已经是10个了
我要涨到12 15 18 20
吞吐甚至没有提升
还有可能会下降
可能是什么原因我们可以想象一下
意味着你的不同的broker上的
partition数量有区别的
有一些broker负责两个 有些负责一个
你这样的这种情况下
producer无法把这些数据
均匀的发到不同的broker上
因为你有些broker的partition数量变多
因为我之前上游的发送策略是
partition是均匀的
这就导致了我们的broker的负载不同
它其实是无法最大化
最大化去保证吞吐能力
从Kafka的存储原理上
其实也是面试中经常会考的
就是说Kafka为什么性能好
因为Kafka的消息文件
就是Kafka的消息
它是存储于我们的Linux文件系统之上的
Kafka它本质上是高度依赖文件系统
来存储和缓存这些消息的
我们一般大家的感知就是我们的HDD
我们的SATA盘可能会非常慢
SSD非常快
Kafka核心还利用操作系统的一些机制
操作系统会将我们的内存有剩余的部分
用来做磁盘的一个缓存
所有磁盘的读写
操作系统都会经过磁盘的缓存
我们也叫什么
我们也叫Buffer IO
对于Linux来说它是个Buffer IO
它会经过一个Linux自己的缓冲
那么什么叫Direct IO
其实就是叫bypass kernel
就是绕开内核
直接读到了我的底层硬件存储
完全不经过我们Linux的page cache
我们在Linux上使用free -m的话
我们看到的所谓的buffer和cache
其实就是一个用于读 一个用于写的
这两个大家是可以去了解一下的
Kafka核心就是用到了这几个特性
第一就是顺序IO
因为我的消息是只追加的
我不像用MySQL
它可能是插到任意位置
就像RocksDB使用LSM Tree
为什么说它的写性能非常好
它其实也是一样的
它核心就是用顺序IO
Kafka也是一样的
就是在文件的尾部去追加这些消息
这是第一个它的顺序IO
因为我知道机械硬盘包括像我们的SSD
它的顺序IO都会比随机IO要好 为什么
对于机械硬盘来说
需要滑动我的机械臂
寻道某一个位置的时候还要等它磁盘旋转
所以它有个寻道延迟和旋转延迟
还有读取的一个延迟
这些都会最终影响我的磁盘性能
你如果是随机IO
每次我机械臂要转来转去
还要等这个旋转时间
肯定是浪费的时间就比较长
但如果你是个连续IO的话
我们硬盘在转的时候 它是连续读取的
所以这样的话性能是最好的
也是最大化IO
第二个就是利用Linux的page cache的能力
因为你刚写的这些数据可能是被我们的
Linux的page cache进行缓冲了
那么再去读的时候
可以先命中到我们Linux page cache
那么它不会去触发IO
所以吞吐也可以最大化
任何发布到partition的消息
就这么被追到尾部了
所以我们看一下这个Topic的写操作
到某一个时间线来说
它可能会被打散到任意一个partition
因为我们partition
分布在不同的broker上
所以它的吞吐能力大家可想而知
其实就是broker
最终约等于
broker的最大的一个顺序IO的能力
这是理想情况下
它约等于全集群的最大的IO吞吐能力
这个是它的一个存储原理
那么既然Kafka集群是使用这种追加写
来保留所有发布的message
那么这个message只要被持久化以后
不管你有没有消费
Kafka是无法直接删除的
因为你可以想象一下你的消息
挨着这样顺序追加写
你要去删除中间这个位置好删除吗
不好删除
因为你删除之后你的文件怎么压缩
没法合并 你从中间删一些数据怎么处理
你文件又不可能把中间删掉
它会自动合并的 这个做不到的
有没有别的策略
Kafka其实是提供了一种叫可配置的保留策略
来去删除旧的一些数据
比方说可以根据保留多少个文件
或者说保留多少天
这些都是常用的策略
比方说我设置的是保留两天
那么在message写入两天后
在两天内它是可以用于消费的
那么之后它将会被丢弃腾出空间
那么Kafka的性能和存储的数据量大小无关
所以数据存储很长一段时间其实也是没关系的
我们怎么去进行文件的丢弃
其实你可以按什么
比方说我可以将Kafka的这种partition
切割成不同大小的segment
按大小去丢弃数据
我也可以按时间
比方我知道每一个segment有个创建时间
我大概就知道哪些可以被安全地删除
另外一个概念叫offset
offset指的是Kafka的一个消息偏移量
我们看这个图里面
这个地方有一个offset等于9
可能是第9条消息
那么这个时候
它产生一条消息 再产生一条消息
所以这个是9 这个10 这个是11
所以我们再看看consumer
Consumer B拿到这个消息可能是offset=11
那么这个时候我们就知道了
offset表示的是每条消息在当前partition下
它是一个唯一的64字节的offset
它是相当于当前分区
第一条消息的偏移量
也就是说表示的是第几条消息
比方说我partition=3
offset=500
意味着什么
意味着500是这个patition 3的第500条消息
这个就是offset
说明这是第几条消息
我们Kafka的消费者
可以指定消费的任意的一个位置
那么这样的话 这种玩法有很多用处
假设我的Kafka消息存储的时间非常长了
那么我可以用于什么
比方做机器学习的时候
我在做AI工程里面
我可以重置一个offset
去重新进行一些数据集的
重新的一些训练
样本的一些重新处理
这是第一
第二个就是说我可以用过去的
一些Kafka消息去做一些审计
做一些别的一些事情
它非常方便 因为不会被丢
唯一的丢弃策略它不是消息级丢弃
它都是文件级丢弃
所以一个是基于大小 一个基于时间
我们再来看一下我们举个例子
假设这个时候我们Kafka集群只有一个broker
会创建两个topic
分别名为topic1和topic2
partition数量分别为1 2
这个时候我们就发现
在Kafka集群配的根目录下
它会创建三个文件夹
topic1-0 topic2-0
topic2-1
也就是我刚刚说的
Topic 1是一个partition
Topic 2是两个patition
那么在Kafka的文件存储中
同一个topic下有多个不同的partition
每个partition都会为一个独立的目录
我们看一下topic1-0
topic2-0 它是个独立的目录
目录里面的partition我刚刚说了
它会被分割成大小相等的segment file
比方说是512MB一个segment file
就是消息按512MB切割
那么segment file
这些文件由两种类型的文件组成
一个就是原始数据 我们叫data file
一种叫index file
index file其实是指索引了
它们总是成对出现
它们的后缀是有些区别
一个是.index结尾
一个是.log结尾
分别表示就是segment的索引文件和数据文件
我们再来看它索引文件它到底是个什么原理
我们看一下这个名字
00好长好长
然后368769.index
我们找到3.497
也就是这个消息第二条
依次在数据文件中表示三个message
前面的3表示的是第三个message
那么它在全局的message排名是多少
它是把368769加上3
368769怎么来的
是索引的文件名称
所以我们就知道了索引文件命名方式是什么
它其实是以上一个文件的最大的
一个消息作为文件名来命名的
所以3表示的在全集群上是368769+3
表示全集群它有多少个
它是第多少个消息
那么497指的是什么
指的是最终log
就是log file它的起始的一个位置
offset等于3的这条记录
它在硬盘上是存在497这个偏移上的
只要打开这个文件
把它seek到这个位置
你就能读取到完整的内容了
所以我们注意index文件它也不是以零开始的
我们看它是以1开始的
1368
而且是非连续的
这种非连续的索引
不是按这种递增一的方式叫稀疏索引
那么这种方式
每隔一定的字节存储一条索引
也就是我们看它中间隔了个2
这里隔了45
这样的话可以减少索引文件大小
使得能够把index映射成内存
降低查询磁盘的IO开销
也就是你这个文件越小
我去解析它
搜索检索它都会变得更快
同时也没有给查询带来太多的时间消耗
因为其文件名是
上一个segment最后一条消息的offset
所以我们需要查找指定offset去消费的时候
它需要去执行一个二分查找
也就是说我捞到一堆这种文件名
它每一个文件名你看0~36879
然后再到7373
这些其实表示的是每一个message的
offset的最大的id的开始的文件名称
我要去找offset怎么找
其实我们大家知道按这个文件名排序
是不是就可以做二分查找了
把这个数字放到内存里面
你们做个二分查找
就可以快速定位到哪个文件
那么如果一旦你快速定位到这个某个文件
你就可以去查index表
快速的去找到这条消息
所以就是它的一个快速检索的能力
Kafka又是如何知道message的偏移呢
Kafka内部的结构
其实message也是像我们之前讲过goim一样
类似有一个这样的一个不同的字段
然后表示大小
这样的话
我可以快速的去读下一个消息跳就好了
因为8 byte offset
然后他还有个什么key length
payload length
都是方便我们做快速跳过这个数据的
我们再总结一下offset表示的是
这个message在当前partition的偏移量
它是一个逻辑上的值
唯一确定的partition中的一条message
它确定了某一个partition的一个
唯一的一条message
可以简单认为是一个id
另外就是
MessageSize就是表示的
是message的data内容大小
data就是具体内容
所以我们看这个地方有个message的
一些字段和它的一些内部的结构
我们再来举个例子
就是我们读取offset=368776的message
我如何快速找到它
我们看一下它分几步
第一步查找segment file
如上图所示就是这个segment file
我们刚才说的
你可以把他的文件名捞出来 做一个什么
做一个二分查找对吧
我找到这个文件名以后
肯定是要先排序 然后二分查找
比方说我就找到了368769
我结果找到了符合它的第一个条件
就是368776离它最近
小于等于它 找到这个文件
第二步我一旦文件找到
因为我们二分查找其实就比较简单了
找到这个segment file之后查找message
通过第一步定位到log file
比方说我的offset是这个
我可以快速的在这个文件里面进行搜索
这个地方我们要看源码
我猜测可以用二分查找
甚至是你顺序扫可能也问题不大
因为它是个稀疏索引
本身也不会特别大
我找到这个offset以后发现
定位到它一些别的信息
定位到什么
就是你可以通过
index文件找到他的文件的偏移
你再去找到这个message
内部的一些内容就全部找到了
所以它是这样快速去搜索出来的
所以说先是通过offset的二分查找找文件名
然后再去打开一个index文件
找到它的偏移
找到偏移之后再打开log文件找到它的具体内容
那么具体的message内容
我们可以按前面的方式去把它解析出来
这个就是它的核心的做法一个思路
Kafka从0.10.0.0版本起
为分片日志文件
新加了一个.timeindex
顾名思义就是可以按照时间戳去检索的
那么它的存储结构其实跟我们的稀疏索引
应该是差不多类似的
你可以通过kafka-dump-log.sh
去查看时间的一个文件内容
首先也是一样的
先定位分片 就根据你时间范围
找到一个最近的日志分段
定位到索引文件
它是通过timeindex
找到它在index中的位置
然后再去找它的文件的一个偏移
最终重复上述的步骤
你就能找到它的具体的一个内容了
这个是它的一个做法
所以timeindex一个是存时间戳
另外一个就是存index文件的offset的id
它是这样来查找的
这一块我们今天先到这里了
我们后面马上就要展开讲它的
producer和consumer
还包括后面leader和follower