我们接着讲producer和consumer
producer就是Kafka的生产者
consumer就是Kafka的消费者
这两个角色我们再看一下在Kafka的设计里面
它是如何去协同协作的
先看producer
producer把消息要发送给Kafka server
也就是broker
我们之前也介绍过 它会根据一个机制
我有好多个partition
我到底应该用一个什么样的算法
把我的某一个message发到
某一个partition
那么如果partition的机制设置的非常合理的话
它是可以尽可能均匀分布到不同的partition的
这里其实就间接的实现了一个负载均衡
我们可以展开讲
就是在Kafka的默认的SDK策略里面
如果你的partition没有设置key
那么第一次调用它会随机生成一个整数
那么后面每次调用就在整数上去自增
那么将这个值与可用的分区数
也就是与partition求余
就可以得到一个partition的值
也就是说最终你求出partition的值
我就写到 partition
把message发到这个partition
有点像我们的一个什么
Round-Robin一个轮询的算法
这种是一个常见的策略
还有一种就是我明确想要指定某一个key
比方我用
最常见的比方说我们希望
某一个用户维度的数据
全部命中到某一个partition
我可能会用一个算法计算用户ID
求余命中某一个partition
这也是一种很常见的做法
这里面有很多种玩法
完全是要看你的业务维度
是不是要某个用户的数据串行
举个例子 我们之前介绍评论的时候
实际上我们是按照评论主题
让它命中到某一个partition
这样的一个做法会不会出现partition不均衡
老实讲是可能会存在的 为什么
因为如果某一个主题的评论特别热
特别爆
那么它可能会命中到某一个partition
这个时候会产生一个partition级别的热点
这个问题不是特别好去解决
因为我们所谓的存在一个数据的倾斜
你的业务的要求上不允许
让它放在三个partition上同时去处理的话
可能会存在一个热点
当然了
也不是说架构上设计上完全没有办法解决
比方说我允许超热的一些业务的数据
当它需要命中某一partition产生热点的时候
我需要一种算法让它均衡地分摊在
partition维度就是可能我有10个partition
我10个partition都有这个维度的一些数据
那么这个时候因为你产生了一个并行
并行度是10
一个partition一个消费者
那么这时候产生了一个锁冲突
打到我的下游的存储
其实也就是锁冲突等于10
我理解这种场景下
相对来说还是可接受的
所以这个时候你就要在存储层面
比方说通过事务的隔离级别
或者说通过我们存储引擎支持的一些这种
锁粒度 锁机制 事物级别
去解决这个问题
也不是说完全没解
这种时候很容易出现这种热点的业务数据的话
你就要考虑牺牲掉局部串行的特性
那变成一个什么
变成一个全局乱序的做法
这是一种负载均衡的策略
如果再展开讲的话
在我们做数据传输
做日志传输
做埋点事件大数据的一些数据传输
因为可能两个事件之间
很难说有很强的一个顺序的保障
这个其实是比较难做到
之前我们是讲微服务可用性的时候
讲过负载均衡策略
我们当然在这个里面
也可以用类似的负载均衡算法
比方说有10个partition
有一个partition特别慢
我这时候一批数据灌过去
我的整个处理逻辑一定
是写一批数据再处理下一批数据
如果有一个partition慢
我刚好命中到这个partition
是不是我的整个吞吐就下降了
我们是不是可以考虑基于partition负载的情况
比方它的写延迟
来决定倾斜一些流量到别的partition
这样的话我的整体的吞吐就变好了
所以在我们内部的话
这种面向日志传输的时候
我们其实会用一些反馈的算法
也是一个负载均衡算法
来倾斜一些流量
就是减少一些流量
到写的慢的partition上
这是一种很常见的一种策略
更极端情况下
如果partition发现一段时间写的特别糟糕
写的特别慢
我们甚至会临时地摘除掉这个partition 完全不写
那么我们把这种机制叫做
partition的黑名单机制
所以你可以看得出来Kafka的这种partition
机制可以让你的玩法是非常丰富的
你可以自己去设定一些负载均衡的策略
决定你的数据怎么去落分区
为了保障Kafka的发送数据
可靠地发送到每个topic
topic的每个partition
在收到producer发送数据后
其实肯定是要ACK的
就是我发送到你的broker
你肯定告诉我要成功还是失败
所以说producer发送成功它会告诉你ACK
就是告诉你到底是成功还是失败
这个是很常见的
在我们的broker收到消息后
如果成功写入落盘了
则返回一个包含消息的主题
写的哪个分区 写的哪个offset
一个叫RecordMetadata的对象
如果失败的话可能就异常了
所以说写到broker成功还是失败
它会返回一个ACK
那么这个ACK其实就会带一个叫
RecordMetadata的对象
告诉你上一次写成功的时候对象
分区offset的信息
这个就是你producer成功的一个回馈
一个ACK
0.11版本的Kafka就很早期的版本
它引入了一个幂等性的做法
幂等性是什么
就是我重复发数据
我们可以想象一下
producer的SDK如果发送到broker失败
你是不是会进行重试 保证数据不丢
重试是不是数据就有可能重复
因为你发给broker
broker虽然能告诉你异常
有可能是因为网络异常
那么网络异常你再次发送的时候
可能数据就重复了
实际上你上一条已经落盘了
所以引入幂等性我们看一下它怎么做的
在producer参数中把这个参数
设为true就可以了 打开
那么开启幂等性的producer在初始化的时候
会分配一个PID
所谓的叫producer ID
那么发往同一个partition的消息
会附带一个sequence number
我们之前在重试的业务体系里面介绍幂等
是不是我们一般所谓的叫流水号
分配一个流水号
那么很显然你也想象的到
在broker端肯定会针对PID
还有partition还有sequence number
这样的一个组合做一个缓存
那么具有相同的主键的这种消息在提交的时候
我的broker可其实只会持久化一次
那么下一个你再提交过来我就跳过它
但是PID因为随着你的producer重启它会变化
所以说不同partition会具有不同的主键
重启的时候肯定是不能保证幂等的
另外幂等性是无法保证跨不同的partition
就是跨分区的exactly once
这个其实是比较难做到的
Kafka的幂等我是不怎么建议使用的
因为它的确是有很多缺陷
我仍然建议你在消费端去做业务的幂等
你想依赖Kafka的幂等去
彻底解决exactly once是非常困难的
我后面会介绍一种机制
怎么去用exactly once去解决传输的一个幂等
好 我们先看一下 consumer消费端
假设有个场景
我们从Kafka中读取消息
写入的消息速度
比我的消费的读取速度快怎么办
因为假设我只有一个消费者
肯定是你很想把consumer的节点
扩多一点
因为你如果不扩的话
随着你的时间流逝
你的消息越堆越多
你只能去通过横向扩展多个消费者
来进行一个水平的扩展
Kafka的消费者是以消费组为一部分的
也就是说我在消费的时候要指定一个消费组
叫Group1 Group2 Group3
那么当多个消费者
形成一个消费组的主题的时候
每个消费者会收到不同分区的消息
比方说我现在有个consumer group 1
我简写为G1
有这么一个主题叫T1
对假设现在有四个分区
如果我现在只有一个消费者
那么这个时候G1的consumer 1
它会收到这4个partition的数据
但是当我新增一个consumer2
就是新起一个节点
那么它指定的参数也是等于G1
也是叫consumer group 1
那么这个时候我们看到这个图
它有个均衡的策略
consumer 1可能消费0和2
consumer 2可能消费1和3
其实就间接实现了一个什么
一个负载均衡的策略
相当于T1的topic里面的partition
均分给了G1中的C1消费者和C2消费者
这个是一个均衡的策略
当我增加到四个并行度四个消费者的时候
我的consumer group 1里面的
所有分区可以刚好一人负责一个
它可以满载去运行
当我继续去增加消费者的时候
其实consumer 5它不会做任何的处理
也就是说我们主题的分区数量
决定了你的并行度
那么我们的问题来了
当你发现你的并行度不够的情况下
肯定会想到我是不是扩partition
假设我只有10个broker
我原则上我建10个partition
可能是已经是最好的结果了
就相当于
每一个broker刚好负责一个partition
因为落盘嘛
我再去建20个是不是
我们随着partition数量越来越多
我的吞吐力未必是线性增长的
我们前面的课件里面也讲了
随着你partition的数量越来越多
未必你的吞吐会增加 为什么
因为有可能从顺序lO退化到了随机lO
相当于你一个目录上有两个partition
可能就是会产生一些IO的争用
这个肯定是不好的
在这种无法大规模去扩partition的情况下 怎么办
我的C1 C2 C3 C4
仍然处理不过来
有没有一个很好的办法和优化
我们抛一个问题
现在大家可以自己想一想
我们是不是可以在consumer 1里面
我进行一个二次sharding
就是我这边收到消息以后
我再做二次sharding
我开多线程
因为之前你一对一消费
我肯定是单线程消费
因为我如果要求局部有序
我肯定是处理完一个再处理下一个
要求局部有序的情况下
你的吞吐量就上不去
如果我在内部做一个二次sharding
拿到一个消息之后分发给不同的线程处理
相当于我在内部开多个goroutine的worker进行消费
那么这个时候其实吞吐是最大化的
但是这里面有个很致命的问题
你的Kafka提交就变得非常困难了
你的Kafka的offset提交是比较困难的
因为你的各个线程的处理进度稍微是有些区别的
因为你offset是递增的
你怎么去提交offset
原则上我们应该是使用watermark机制
整个你的线程池里面处理的
最慢的offset才能被提交
你可以想下这个代码应该怎么写
那么这样的话最小的offset提交的话
才能保证你的二次sharding多线程的
多goroutine的一个消费 它才能正确
这个代码你可以想象一下怎么做
还有一个问题是什么呢
Kafka的一个很重要的特性就是写入一次消息
可以被n个consumer group同时消费
也是我最喜欢的一个功能
就是你可以开consumer group 1
你是不是可以开个consumer group 2
那么开两个不同的consumer group
意味着partition里面的这些数据
可以被镜像消费 举个实际例子
我们开一个consumer group 1
消费数据写到HDFS
再开个consumer group 2写到Elasticsearch
那么大家可以全量去消费同等的数据
在我们业务形态里面这个功能也非常好用
这样的话我的业务的很多微服务
通过区分不同的consumer group
不就是可以解决不同的全量消费的问题嘛
这个其实是非常好用的
当然了如果你的两个consumer group
消费速度差异特别大
其实会间接导致Kafka的性能衰减
这个时候问题怎么出现的
我们前面讲Kafka它的最大的优势
在于顺序IO和page cache的使用
如果你只有一个consumer group
刚生产立马消费
因为你的消费速度很快
这种情况下可以充分利用page cache
但是如果你新增了一个consumer group
那么这个consumer group可能消费能力比较差
会不会导致
一个生产速度和消费速度能匹配上追得很快
另外一个消费特别慢
导致page cache污染
因为你去读数据的时候会用page cache的buffer
但是你想想你读的很老的那些数据
还仍然使用page cache
是不是有可能
会跟我当前写的速度非常快的那些page cache
因为内存不是无限大的
它会产生一些争用
那么这时候就会产生一些性能衰减
所以我们有时候会讲超大规模Kafka集群
可能会存在这些性能瓶颈
这样的一些特殊的case导致的
这个有没有办法优化
其实也有 你可以想象一下
是不是说当我的新增的consumer group
跟我最快的consumer group
它们的接近的时候
我再开启page cache
那么在两个offset差别特别大的时候
我不开启预读 我直接走direct IO
我绕过Linux的buffer IO
这个就是我们所谓的direct IO
这样其实也是可以解决这个问题的
这个可能要在Kafka的底层去优化这个问题
我们再来看
还有一个问题
我们在前面讲了
大家可能会有疑问 就是我的consumer
在加入的过程中如何自动的去做负载均衡
新加一个consumer
如何让这个partition可以很好地均分给我的consumer
大家可以关注一下早期ZooKeeper早期Kafka的版本
如何去实现partition的一个均分
其实早期是用的ZooKeeper
也就是说我的消费者注册到ZooKeeper
它注册的是一个临时的顺序节点
这样的话我的consumer其实是
知道全局一共有多少个consumer
那么把我consumer排排序
然后我又知道一共有多少个partition
不就很好的均分了吗
大家得的顺序都是一致的
那么新的Kafka的版本它就去掉了ZooKeeper的依赖
之前那个做法我建议大家也是去学习参考一下
有些时候利用etcd或者
ZooKeeper可以做到类似的一些事情
你可以参考一下它的一个玩法
那么新的的时候它怎么做
重新均衡Kafka是一个很重要的性质
因为这样的话在保证新加入或者退出consumer
它可以重新被均衡
怎么做
消费者呢定期发送心跳Heartbeat
作为一个组协调者叫Group Coordinator
那么这个Group Coordinator
其实它是个特殊的broker
通过一个心跳知道哪些consumer是存活的
就以前依赖ZooKeeper我现在不依赖了
现在依赖选一个broker来做这个事情
选一个组协调者
那么这个broker不是固定的
每个消费组可能是不同的
这样的话可以负载均衡
相当于每一个broker可能负责group1
另外一个broker负责group2
这样的话是很好的一个机制
那么这个时候当消费者拉取消息
或者提交的时候便会发送心跳
那么如果消费者超过一定时间没有发生心跳
认为会话过期可能就认为它挂了
组协调者会认为它宕机
就会触发重新平衡
因为你有时候重启的时候
或者我扩容的时候
都会导致分区就是消费者去变更
所以这个机制其实是个动态很好的一个机制
那么也就是可以看到
消费者宕机到会话过期
是有一定时间的
那么这段时间内
消费者的某一个分区是不能进行消息消费的
因为它挂掉了
你要感知到重新均衡
你才能重新消费这个分区
那么通常情况下
我们可以更好的处理这个问题
就可以优雅关闭
消费者主动离开的时候告诉你
我要主动下线
这样的话可以立即触发重新均衡
那么在0.10.1版本Kafka
对心跳机制还进行了一些优化
将消息的拉取和心跳剥离了
就是心跳的频率不受拉取的频率影响
避免相互影响
另外更高版本的Kafka支持一个消费者
多长时间内不拉取消息
仍然保持存活
这个什么意思呢
避免活锁 你可以想象一下
有一个消费者它长时间不来拉取
拉取Kafka的一个消息
我们为什么叫活锁
因为这种情况下虽然它的心跳是健康的
但是因为它的程序可能出了一些Bug
死锁了或者是卡住了
导致这个分区的消息一直不被消费
你这个肯定是业务就会受影响
所以说这种情况下活锁可能会导致
某一个分区数据丢失或者说数据延期
这样的话下游的组件性能退化 怎么办
所以Kafka支持这个策略叫什么
有多长时间不来拉消息
我就会自动触发rebalance
这个策略
可以很好地检测出处于活锁的这些消费者
因为它心跳正常
但是它不消费或者消费能力特别差
它可以检测出来
但是正也是因为这个功能其实可以
检测出这些有问题的节点触发重均衡
一定要注意这个机制其实是有那么一定风险的
我们之前就踩过这个坑
我们之前在做Kafka消费日志 消费事件的时候
利用活锁机制导致频繁触发rebalance
因为rebalance在你分区数量很多
消费者数量很多时候
rebalance是需要时间的
那么rebalance期间因为数据吞吐它是会陡降的
所以整个吞吐能力会下降的非常厉害
我们当时是怎么出现这个问题的
可以讲一下
当我的下游HDFS吞吐特别慢
写的特别慢的时候
是不是很容易触发活锁的检测
导致触发rebalance
触发rebalance就进一步导致我的吞吐下降
导致我的消息堆积
所以它就进入恶性循环
写HDFS慢 触发活锁
触发活锁然后重新均衡
吞吐更加下降 消息更加堆积
当我均衡完之后重新写HDFS
又进入到一个活锁
其实它就进行了出现了这么一个问题
当然我们后面这个问题怎么解决
是通过指定partition消费来解决的
就是我的Flink任务指定partition来消费
避免说用自动rebalance这个机制
这个也是我们之前踩过的一个坑
就告诉一下大家
partition以前早期的partition
它的consumer group的消费进度
是存储在ZooKeeper的 就是很早期的
那么现在新的版本这个机制要优化了
因为我们知道
依赖ZooKeeper这种东西其实是比较重的
让我们的客户端SDK变得非常复杂
现在其实是会有一个0.9开始啊0.9这个版本开始
将消费端的一个offset信息存储在集群内部
也就是在Kafka内部搞定了这个问题
它的做法比较巧妙
它新建了一个叫做__consumer_offsets的主题
它默认是50个分区
然后里面的写的数据其实就是写了一个
TopicPartition.OffsetAndMetadata
存储了这个数据
那么key其实主要就是
存放主题分区以及消费组信息的
那么value就是offset和metadata
包括位移 offset offset提交时间
自定义的一些metadata
这样的话我就不需要依赖ZooKeeper了
那么你可以想象一下你的offset一直在提交
比方说partition 1 T1
它提交的offset 0
第二次提交1到3
那么其实0123你提交完之后
是不是只有最后一条数据有效
所以Kafka又支持一个叫partition级别的压缩
只保留最后一条数据
那么这样的话你看
就可以压缩数据就变得比较小
我们再来看一下
整个consumer的commit offset
就是提交offset的时候它的一些机制
那么消费端就是在我们consumer的SDK
可以设置参数叫enable.auto.commit
来控制手自动提交还是手动提交
如果为true它就自动提交
那么在消费端后台
它会定时把offset给提交上去
你可以自己去配这个时间
比方说默认是5秒钟
那么这样的一个配置
可能会导致什么问题呢
可能存在重复的位移数据提交
因为每隔5秒钟去写一条消息
可能会产生大量相同的key消息
因为你其实只需要最后一条
这时候offset没有推进的时候你会重复提交
当然了我们刚才说的可以依赖我们的
partition的日志压缩来清理这些数据
第二个叫重复消费
假设你的位移提交时间是5秒钟
这5秒内如果产生了一个rebalance
那么所有的消息
会从上一次提交的offset进行一个消费
那么这个期间消费的数据会被再次消费
你可能这5秒钟可能有些重复数据
我们再来看一下Kafka的这种就是
由消费端来提交offset的这种机制
其实有两种玩法
第一种就是说我读完消息
先commit再处理消息
这种模式下如果consumer在commit后
还没来得及处理就crash掉了
这个时候你在下次重启的时候
因为你是从上一次offset去拉取消息的
这个时候消息就丢了
所以我们叫什么
它叫至多一次消费
它至多只会消费一次
但是可能会丢失
那么第二种模式叫做
读完消息我是先处理这个消息
然后再commit offset
那么这种模式下
如果在处理完消息之后
commit之前consumer挂掉了
那么下次重新开始的时候
还是会拉到刚刚未commit的消息
实际上这个消息已经被我处理过了
所以我们叫什么叫at least once
它是至少一次消费
这种时候
就要求我们的业务去处理消息的幂等
这个其实是挺关键的一个事情
那么这个是Kafka的两套机制
但是注意有一些场景结合二阶段提交
就能做到所谓的exactly once
我这里举一个场景
举个例子 在我们Flink
就是大数据的一个实时计算平台
大家如果对Flink不了解
其实也可以去简单了解一下
因为我之前提过很多次Flink
其实Flink的整个机制叫什么
叫二阶段提交
我们先看它的核心的原理
比方说我有个源
被摄取读出来的一些数据
读出来的数据我可能要写到一个地方去
我可能做些业务逻辑处理可能是个写操作
如果把这个写操作
直接更新到你的结果的目标库
我是不是如果Kafka
重新重启的时候或者故障的时候
重新拉到这些消息再去写的时候
是不是就把以前的正式数据给搞脏了
那么什么是二阶段提交
就是说我读出来这个数据先写到临时的地方
写到临时的地方之后
我再验证完它可以被提交了
然后我再把变更应用到我的destination
就是我的最终的应该被处理的地方
这样的一个二阶段机制
结合我们的Kafka的offset同时保存
是不是可以做到一个所谓的exactly once
我们看一下Flink里面它怎么做到的
第一步从Kafka消费数据写入到一个临时目录
注意它是个临时目录
就正式环境不可见
在Fink的checkpoint
什么是checkpoint
就是以前我打游戏
每次到关卡里面要打BOSS前
基本上都会自动存档
为什么要存档
因为你要挂 你可能打BOSS打挂了
你的存档记录就丢失了
你不就回滚了
又回到你之前上一次存档的地方
你就白打了
同样的Flink其实也是有存档的
那么在存档的快照阶段
就是因为你消费到哪
我要把进度保存到
Fink的state中 这个是个存储
存储到state中
同时在sink端立马关闭之前写入的那些文件
以及保存我的存档的id到state中
也就说我在这个state中存了两个东西
checkpoint的id
checkpoint就是我的存档的存档号
还有我的Kafka的offset
这两个同时存储成功
是不是意味着我在挂掉的时候
我再拉起这个任务
是不是这两个可以同时还原回来
所以说这是一个快照阶段
那么在存档的complete阶段
先commit Kafka offset
我这批数据写完了
commit Kafka offset
同时将这些临时目录移到正式目录
第三步就是所谓的二阶段提交
我先提交Kafka offset同时挪到正式目录
有没有可能在commit Kafka offset的时候
挪正式目录的时候失败 没关系
你不需要回滚
你可以等到下一次
再来做checkpoint commit的时候
或者在我们checkpoint recover的阶段
你要读档的时候
你可以再做一次补偿
我们看它怎么补偿
先恢复state的信息
state里面存的啥
先reset Kafka offset
你要从这个位置重新去消费
然后恢复last checkpoint id
你的last checkpoint id也找到了
找到这个之后你是可以找
比last checkpoint id要小的那些目录
临时目录挪到正式目录
因为有可能是上一次它没有挪成功
残存在这个地方
我就尝试把它挪过去
那么这样的话其实数据也是会最终一致的
这个我们在内部用Fink结合Kafka
做到了所谓的exactly once
其实非常巧妙的一个做法
二阶段提交先写临时的
我们再来看看consumer的它的
push和pull的一个策略
那么作为一个消息系统
Kafka遵循了传统的方式
Kafka的producer肯定是
向broker去push消息的 肯定是推的
consumer从broker它其实是pull消息
也就是它是去拖消息
那么在一些这种日志的系统
比方说像Facebook的Scribe
像Flume它们基本上都是push的做法
但是事实上push和pull它各有优劣
我们来分析一下
push模式它有个缺陷
很难适应消费速率不同的消费者
也就是说有些人消费的快
有些人消费的慢
如果你是无脑push的话
是不是没必要
因为有些人处理的慢
我其实要push得慢一些
那么消费发送数据
消费者的发送数据是broker决定的
那么push模式的目标肯定是
尽可能最快的速度传递到下游
传递消息 那么很容易导致
consumer来不及处理
你这个时候肯定是要做一些机制
比方说发现你的内存快满了
消息队列过长了
反压给我的broker
典型的就是说我要拒绝服务了
或者是网络拥塞
或者我HDFS写的特别慢
你这个时候就做了一些无用功
你要反压回去
pull模式相反
它可以很好地根据consumer的消费能力
以适当的速率去消费消息
因为你消费的慢你就拖的慢
所以它是可以适配消费速率的
还有对于Kafka而言
pull模式可能更合适 为什么
因为pull模式可以简化broker的设计
也就是说consumer可以自主去控制消费消息的速率
我不是说push做不到消费速率的一个适应
比方说我之前也介绍过
像RabbitMQ它是这种push策略的
在consumer端用一个类似的
监控这个消息队列
我push给你 不是说立马就消费得完
我在consumer有个队列
它可以监控这个队列长度
然后自动地做反压
这个会导致整个代码其实会变得更复杂
那么consumer这个做法就比较简单了
就完全给consumer自己去拖
拖多少消费多少
你拖得慢
就每次来同步来拉的时候拉的慢
那么同时可以让consumer
自己来考commit offset
那么这样的话也可以比较好的
实现at most once
和at least once两个不同的提交语义
这个其实是Kafka最有魅力的一个点
那么pull模式还可以根据
consumer的消费能力适当的去消费消息
但是pull模式它的不足之处就是说
如果Kafka没有数据
它是不是会轮询
没事来看一下有没有数据
这时候会浪费一些CPU
因为你每次拿不到数据返回空数据
那么针对这一点其实也比较好优化
那么消费者从broker去主动拉数据的时候
它可以做一个长轮询
长轮询其实我们在goim的时候也介绍过
我们一般是长连接
但是也可以长轮询
比方说像Discovery
Eureka之前这种轮询的方式
我们改成长轮询
这一点其实Kafka做法应该是类似的
那么针对这一点Kafka消费
它在消费出去的时候会传入一个时长timeout
也就是它不会无限地傻等
它会等一段时间直到timeout
然后再做下一次轮询
所以它是一个长轮询的机制
那么关于producer和consumer
一些这些知识我们就基本掌握了
所以我们总结下来就是说
你一定要想清楚consumer的消费语义
以及业务的消费幂等