kafka理论
消息队列作用、模式
作用:1.消峰 2.解耦 3.异步通信
模式:1.点对点模式(删除对应的消息,只有一个消费者)
2.发布订阅模式(不删除数据 消费者相互独立 可以多个消费者)
kafka中概念
Producer(生产者) 分区(一个topic分为多个分区,) Consumer(消费者)
leader,follower:生产消费主要针对leader,leader挂了follower上线
kafka为什么快
1.分布式集群,采用分区 2.稀疏索引,快速定位
3.顺序写磁盘(600M) 4.页缓存和零拷贝(零拷贝:数据传输不走应用层)
生产者分区
分区的好处:1.提高并行度 2.便于合理的存储
分区的策略:
1.指明partition时,直接用
2.没有指明partition时:key的hash值与topic的partition进行取余
3.没有partition,没有key:粘性分区(随机选择一个分区,
一直使用这个分区,直到batch满或者完成)
ZK
ZK中记录的信息:leader信息,分区
ZK中的kafka信息:
1./kafka/brokers/ids 记录哪些服务器
2./kafka/brokers/topics/first/partitions/0/stats 记录谁是leader,哪些服务器可以用
3./kafka/controller 辅助选举Leader
Kafka副本概念
Kafka副本:(副本统称为AR=ISR+OSR)
默认个数1,一般改为2。
操作对象只是针对leader。
ISR:表示和Leader保持同步的Follower集合。
OSR:Follower与Leader副本同步时,延迟过多的副本(超过30S)。
LEO(Log End Offset):每个副本中最后一个offset,LEO其实就是最新的offset+1
HW(High Watermark):所有副本中最小的LEO
副本的调整
分区副本的调整:
auto.leader.rebalance.enable=true
//不平衡的比率 一般不开启
leader.imbalance.per.broker.percentage=10%
leader.imbalance.check.interval.seconds=300s //负载均衡检查时间
增加副本因子:
//副本存储计划
--reassignment-fileincrease-replication-factor.json --execute
Leader选举流程
1.在ZK中注册 2.controller谁先注册,谁输了算
3.选举出来的controller监听broker变化
4.ISR中存活为前提,AR中排在前面的优先 5.controller将节点信息上传到ZK
6.其他controller从ZK同步相关信息
7.broker挂了,controller监听到变化 进行新一轮的选举
故障流程
Follow故障:
1.被踢出ISR,leader和follower继续接受数据
2.将log文件中高于HW的部分截取掉,从HW开始向leader进行同步直至HW。
完成后加入ISR。
Leader故障:
1.ISR选出新的Leader 2.其余follower将自己文件中log高于HW部分截取掉,
然后从新的leader同步数据
只能保证数据的一致性,并不能保证不丢失或者不重复
文件存储机制
1topic==N partition==N segment
.index 偏移量索引文件(稀疏索引4K数据,记录一条数据。offset为相对offset)
.timeindex 时间戳索引文件(判断文件删除)
index和log文件以当前segment的第一条消息的offset命名
上面文件在文件夹下面:文件夹的命名规则为topic+分区
文件清除策略
log.retention.hours 默认7天 log.retention.check.interval.ms 检查周期5分钟
delete策略和compact策略
//最大时间戳作为该文件的最大时间戳
log.cleanup.policy=delete
//对于相同key的不同value,只保留最后一个版本。(适用于用户信息等)
log.cleanup.policy=compact
kafka消费方式
(拉)如果没有数据,可能会陷入循环中,一直返回空数据
一个消费者消费一个分区+的数据。offset 在系统主题中
消费者组内消费者大于分区数:会有空闲的消费者
groupid的hash值 % 50(分区数量)
kafka消费参数
Fetch.min.bytes=1 //每次最小抓取大小
Fetch.max.bytes=50m //一批数据最大值
Fetch.max.wait.ms=500ms //一批数据最小值
Max.poll.records=500 //一次拉取数据返回消息的最大条数
消费者分区
Range、RoundRobinAssignor、Sticky、CooperativeSticky
partition.assignment.strategy=Range+CooperativeSticky
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
"org.apache.kafka.clients.consumer.RoundRobinAssignor")
//(多topic:所有多出来的分区都在1分区)
Range:针对一个topic 分区数/consumer数 前面的消费者多消费
RoundRobin:针对多个topic Range
Sticky:相邻的分区尽量在一起
消费者分区再平衡
触发条件:1.超时45s 2.消费者消费时间过长5min
offset维护位置
__consumer_offsets kafka内置topic
key value存储的:key是group.id+topic+分区号 value就是当前offset值
一段时间进行压缩:group.id+topic+分区号 保留最新的数据
自动提交offset:enable.auto.commit=true auto.commit.interval.ms 提交的时间间隔默认5s
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"true")
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000")
手动提交offset:
同步提交:必须等待offset提交完成 异步提交:offset提交后,就开始消费下一批数据
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"false")
kafkaConsumer.commitAsync()
kafkaConsumer.commitSync()
指定offset:
auto.offset.reset=earliest|latest|none
earliest:自动将偏移量重置为最早的偏移量
latest:最新的偏移量
none:如果未找到先前偏移量,则消费者抛出异常
其他理论
数据积压:
1.增加分区和增加消费者
2.增加拉取单次条数
kafka监控:kafka-Eagle监控
kafka默认内存1G
kafka-kraft模式:去掉ZK
标签:副本,理论,kafka,topic,offset,leader,分区
From: https://www.cnblogs.com/wuxiaolong4/p/16800524.html