Kafka知识点梳理
2023-12-26 18:58:14 27 举报AI智能生成
登录查看完整内容
Kafka是一个分布式流处理平台，由LinkedIn公司开发并开源。它主要用于构建实时数据管道和流应用。Kafka的核心概念包括Producer、Broker、Consumer和Topic。Producer负责生产消息，发送到Broker；Broker接收并存储消息，根据配置的分区规则将消息分发到不同的Topic；Consumer订阅特定的Topic，从Broker拉取消息进行消费。Kafka具有高吞吐量、可扩展性、持久性和容错性等特点，适用于大数据场景下的实时数据处理。
作者其他创作
大纲/内容
在 ZooKeeper 的帮助下管理和协调整个 Kafka 集群
概述
第一个成功创建 /controller 节点的 Broker 会被指定为控制器
选举
主题管理（创建、删除、增加分区）
分区重分配
Preferred 领导者选举
集群成员管理（新增 Broker、Broker 主动关闭、Broker 宕机）
作用
各broker watch /controller临时节点，故障后重新选举
故障转移（Failover）
0.11 版本后，把把多线程的方案改成了单线程加事件队列的方案
将之前同步操作 ZooKeeper 全部改为异步操作
内部设计原理
Controller(控制器)
读写分离不能提高读性能
方便实现“Read-your-writes”
方便实现单调读（Monotonic Reads）
为什么kafka不支持读写分离
问题
使用 __consumer_offsets内部主题的方式来保存位移
实现方式
Value格式：主要是offset的值，还包括时间戳等
Key的格式：<GroupID，主题名，分区号 >
普通消息格式
保存Consumer Group 信息的消息
用于删除 Group 过期位移甚至是删除 Group 的消息
消息格式
如果位移主题是 Kafka 自动创建的，那么该主题的分区数是 50，副本数是 3
默认设置
消息Compaction
__consumer_offsets
Consumer 程序会处于阻塞状态，直到远端的 Broker 返回提交结果
commitSync()
立即返回，不会阻塞，不会影响 TPS，但是有错误不会重试
commitAsync()
直接提交最新一条消息的位移
精细化提交位移
手动提交
enable.auto.commit=trueauto.commit.interval.ms（默认5s）
参数设置
在开始调用 poll 方法时，提交上次 poll 返回的所有消息。导致问题：它可能会出现重复消费
提交位移时机
自动提交
手动提交，组合异步提交和同步提交
最佳实践
CommitFailedException 异常通常发生在手动提交位移时，即用户显式调用 KafkaConsumer.commitSync() 方法时
发生时机
同时出现了设置相同 group.id 值的消费者组程序和独立消费者程序
消息处理的总时间超过预设的 max.poll.interval.ms 参数值
原因
缩短单条消息处理的时间
增加 Consumer 端允许下游系统消费一批消息的最大时长，调大max.poll.interval.ms
减少下游系统一次性消费的消息总数，调小max.poll.records
下游系统使用多线程来加速消费
解决方案
CommitFailedException
提交位移
位移offset
只有一个作用，就是提供冗余以实现高可用
Kafka的读写只能发生在leader副本上
判断标准 replica.lag.time.max.ms（默认10s）
ISR(In Sync Replicas)
Broker 端参数 unclean.leader.election.enable 控制是否允许 Unclean 领导者选举
参数
在这个问题上，Kafka 赋予你选择 C 或 A 的权利
选择
建议
Unclean 领导者选举
副本
=Consumer Offset -LogStartOffset
计算方式
lead越接近于0，那么就表示有可能要丢消息
lead
=HW-Consumer Offset
消费者当前落后于生产者的程度
lag
消费者组监控
监控
新组成员加入或老组成员离开
组成员崩溃，或者网络异常被踢出组
组成员数发生变更
使用正则表达式来订阅topic
订阅主题数发生变更
增加分数区
订阅主题的分区数发生变更
触发时机
在 Rebalance 过程中，TPS不高。所有 Consumer 实例都会停止消费，等待 Rebalance 完成
在Rebalance 时，效率不高。所有 Consumer 实例共同参与，全部重新分配所有分区
Rebalance过程太慢
订阅主题分区的分配由Coordinator负责
所有Broker 都有各自的 Coordinator 组件
第 1 步：确定由位移主题的哪个分区来保存该 Group 数据：partitionId=Math.abs(groupId.hashCode() % offsetsTopicPartitionCount)。
第 2 步：找出该分区 Leader 副本所在的 Broker，该 Broker 即为对应的 Coordinator。
Consumer Group 如何确定为它服务的Coordinator 在哪台 Broker 上?
Coordinator
Consumer 实例会被 Coordinator 错误地认为“已停止”从而被“踢出”Group
可避免情况
每个 Consumer 实例都会定期地向Coordinator 发送心跳请求，表明它还存活着
session.timeout.ms
控制发送心跳请求频率
heartbeat.interval.ms
它限定了 Consumer 端应用程序两次调用 poll 方法的最大时间间隔。它的默认值是 5 分钟
max.poll.interval.ms
相关参数
设置 session.timeout.ms = 6s设置 heartbeat.interval.ms = 2s要保证 Consumer 实例在被判定为“dead”之前，能够发送至少 3 轮的心跳请求
max.poll.interval.ms的值要大于消费者消费的最大时长
方案
如何避免
靠消费者端的心跳线程（Heartbeat Thread）
重平衡过程是如何通知到其他消费者实例的？
过程
Rebalance
设置ack=all
设置retries 次数
生产者
enable.auto.commiot=false，确保消息消费完成再提交
消费者
设置unclean.leader.election.enable = false 分区落后太多不允许竞选为leader
replication.factor >= 3
min.insync.replicas > 1
确保 replication.factor > min.insync.replicas
Broker
消息可靠性
调大 num.replica.fetchers（ Follower 副本用多少个线程来拉取消息）
调优GC参数避免FGC
Broker端
适当调大batch.size的值
适当增大linger.ms的值
设置compression.type为lz4或者zstd
设置ack为0或者1
设置retries为0
多线程共享Producer实例，则适当调大buffer.memory
Producer端
采用多线程消费
增大fetch.min.size值
Consumer端
高吞吐
低延迟
调优目标
禁止atime mount -o noatime
选择合适的文件系统 ext4 或 XFS
swap 空间设置得比较小 sudo sysctl vm.swappiness=N
加大文件句柄数 ulimit -n
预留较大的页缓存
调大 vm.max_map_count
操作系统层
将JVM 堆大小设置成 6～8GB
使用 G1 收集器
大对象 增加 JVM 启动参数 -XX:+G1HeapRegionSize=N
JVM
即尽力保持客户端版本和 Broker 端版本一致
不要频繁地创建 Producer 和 Consumer 对象实例
用完及时关闭
合理利用多线程来改善性能
应用层
调优层次
性能调优
Kafka知识图谱
0 条评论
回复 删除
下一页
