Kafka 3.x 实战运维手册(ZooKeeper 版,示例基线 3.6.2)
版本说明(2026-09-17 校订) :本文命令与配置示例以 Kafka 3.6.2 为基线,适用于仍在维护的存量 ZooKeeper 模式集群。Kafka 3.x 的最终补丁版本为 3.9.2;3.6.2 不是 3.x 终版。Kafka 4.0 已移除 ZooKeeper 模式,新建集群应优先采用 KRaft。本手册不能直接当作 Kafka 3.9 或 4.x 的参数模板使用。
适用版本 :Apache Kafka 3.0.x ~ 3.9.x,元数据存储采用 ZooKeeper(非 KRaft)面向读者 :运维工程师、SRE、中间件平台工程师、后端开发(需要理解 Kafka 运行机制)编写目标 :从零搭建 → 稳定运行 → 性能调优 → 故障自愈 → 容量规划,全链路可落地约定 :所有命令以 Linux(CentOS 7+/Ubuntu 20.04+)为准,Kafka 安装目录统一记作 /opt/kafka,ZooKeeper 数据目录记作 /data/zookeeper,Kafka 数据目录记作 /data/kafka。请根据实际环境替换。
目录
第 0 章 快速开始(10 分钟跑起来)
本章用于理解 Kafka 长什么样 和验证环境 ,生产部署请直接从第 2 章开始。
0.1 单机快速体验(ZooKeeper 模式) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 java -versioncd /opt wget https://archive.apache.org/dist/kafka/3.6.2/kafka_2.13-3.6.2.tgz tar -xzf kafka_2.13-3.6.2.tgzln -s /opt/kafka_2.13-3.6.2 /opt/kafkacd /opt/kafka bin/zookeeper-server-start.sh -daemon config/zookeeper.properties bin/kafka-server-start.sh -daemon config/server.properties bin/kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic quickstart --partitions 3 --replication-factor 1echo -e "hello\nkafka\nworld" | bin/kafka-console-producer.sh \ --bootstrap-server localhost:9092 --topic quickstart bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic quickstart --from-beginning --timeout-ms 5000 bin/kafka-server-stop.sh bin/zookeeper-server-stop.sh
0.2 你该知道的 5 件事
Kafka 3.x 仍支持 ZooKeeper ,Kafka 4.0 已移除 ZooKeeper 模式。3.x 是 ZooKeeper 模式的最后一个大版本,存量集群应规划向 KRaft 迁移。
Kafka 的吞吐来自顺序写、页缓存、批量与零拷贝等机制 。磁盘吞吐和尾延迟仍会直接影响生产请求、副本同步与恢复速度,不能忽略。
分区数一旦增加不可减少 。Kafka 3.x 不支持减少 Topic 分区数。
副本数(replication.factor)决定了容灾能力 ,1 副本 = 无容灾,生产最低 2,推荐 3。
acks、min.insync.replicas、replication.factor 三个参数共同决定可靠性 ,必须一起设计。
第 1 章 Kafka 核心概念与架构 1.1 一句话理解 Kafka Kafka 是一个分布式的、分区的、多副本的提交日志(Commit Log)系统 ,本质上是一个”可以持久存储、可以被多方重复读取、可以水平扩展的消息队列”。
1.2 核心概念对照表
概念
说明
运维关注点
Broker
一个 Kafka 服务进程/节点
单节点承载能力、磁盘、网络
Topic
逻辑消息分类,类似”表”
分区数、副本数、保留策略
Partition
Topic 的物理分片,是并行与顺序的最小单位
数量决定并发度,不可减少
Replica
分区的副本,分为 Leader / Follower
副本数决定容灾能力
Leader
处理所有读写的副本
Leader 分布是否均衡
Follower
只做同步,不对外提供服务
ISR 同步延迟
ISR
In-Sync Replicas,与 Leader 保持同步的副本集合
核心健康指标,ISR 收缩 = 有风险
OSR
Out-of-Sync Replicas,落后过多的副本
长期 OSR 会拖慢 Leader
Controller
集群中负责分区 Leader 选举、元数据管理的 Broker
单点,需关注切换
ZooKeeper
存储集群元数据、Controller 选举、ACL
3.x 强依赖,必须高可用
Offset
消息在分区内的唯一递增编号
消费进度、滞后量
Consumer Group
消费者组,组内分区消费分摊
再平衡(Rebalance)
Segment
分区目录下的日志文件分片
文件数量、清理策略
HW (High Watermark)
消费者可见的最高 offset
影响消费延迟
LEO (Log End Offset)
分区日志末尾 offset
LEO - HW = 副本延迟
LSO (Last Stable Offset)
事务场景下稳定 offset
事务消息排查
1.3 架构图(ZooKeeper 模式) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 ┌──────────────────────────────┐ │ ZooKeeper 集群 │ │ zk1:2181 zk2:2181 zk3:2181 │ │ 元数据 / Controller选举 / ACL │ └──────────────┬───────────────┘ │ ┌──────────────────────────────┼──────────────────────────────┐ │ │ │ ┌───────▼────────┐ ┌────────▼───────┐ ┌─────────▼──────┐ │ Broker 1 │ │ Broker 2 │ │ Broker 3 │ │ (Controller) │◄──────────►│ │◄──────────►│ │ │ │ 副本同步 │ │ 副本同步 │ │ │ P0 -L P1 -F │ │ P0 -F P1 -L │ │ P0 -F P1 -F │ │ P2 -F │ │ P2 -F │ │ P2 -L │ └───────┬────────┘ └────────┬───────┘ └─────────┬──────┘ │ │ │ └──────────────────────────────┼──────────────────────────────┘ │ ┌──────────────────────────────┼──────────────────────────────┐ │ │ │ ┌───────▼────────┐ ┌────────▼───────┐ ┌─────────▼──────┐ │ Producer 集群 │ │ Consumer 集群 │ │ 监控/治理平台 │ │ acks=all │ │ Group A/B/C │ │ JMX/Exporter │ └────────────────┘ └────────────────┘ └────────────────┘
1.4 消息写入流程(必须理解) 1 2 3 4 5 6 7 8 1. Producer 根据 partitioner 决定写入哪个分区 - 指定了 key → hash(key) % numPartitions - 未指定 key → 3.x 之前:round-robin;3.x+:粘性分区(Sticky)批量填充2. 发往该分区 Leader 所在的 Broker3. Leader 写入本地 Log(PageCache → 磁盘)4. Follower 主动拉取(fetch)同步5. 当 ISR 中所有副本都写入成功(acks=all)→ Leader 更新 HW6. Consumer 只能消费到 HW 之前的消息(默认)
1.5 副本同步机制(LEO / HW / ISR) 这是 Kafka 可靠性的核心,运维必须清楚:
1 2 3 4 5 6 7 Partition P0 三个副本: Leader : LEO =100, HW =95 ← 写入最快 Follower1: LEO =98, HW =95 ← 落后 2 条,仍在 ISR Follower2: LEO =60, HW =60 ← 落后 40 条,可能被踢出 ISR HW = min(所有 ISR 副本的 LEO) = 95 Consumer 最多消费到 offset =94
触发 ISR 收缩的条件 (由 replica.lag.time.max.ms 控制,默认 30000ms):
Follower 在 30 秒内没有向 Leader 发起 fetch 请求,或
Follower 的 LEO 落后 Leader 超过 30 秒(注意:是时间 不是条数!3.x 已废弃 replica.lag.max.messages)
ISR 收缩的后果 :
acks=all 的写入只需 ISR 内副本确认 → 变快但可靠性下降
若 ISR 缩到只剩 Leader,此时 Leader 宕机 → 数据丢失
⚠️ 运维红线 :必须配置 min.insync.replicas >= 2,否则 ISR 缩到 1 时 acks=all 形同 acks=1。
1.6 Leader 选举机制
场景
选举方式
是否丢数据
Leader 宕机
从 ISR 中选新 Leader(Controller 主导)
不丢(ISR 内的都同步过)
ISR 全挂 + unclean.leader.election.enable=true(是否可以从没有同步完全的副本当中选择新的leader为true)
从 OSR 中选
会丢数据
ISR 全挂 + unclean.leader.election.enable=false(是否可以从没有同步完全的副本当中选择新的leader为false)
分区不可用,等待原 Leader 恢复
不丢,但不可用
生产建议 :unclean.leader.election.enable=false(默认值),宁可不可用也不丢数据。若业务允许丢数据换可用性(如日志采集),可设为 true,但必须明确告知业务方。
1.7 关键版本特性(3.x 时间线)
版本
重要特性
运维影响
3.0
KRaft 预览、移除旧版客户端协议
无影响(ZK 模式仍可用)
3.1
KRaft 支持 ACL/SCRAM
-
3.2
KRaft 支持动态分区变更
-
3.3
KRaft 生产可用、增量式 AlterPartition
ZK 模式仍主流,可试点 KRaft
3.4
-
-
3.5
KRaft 支持迁移(ZK→KRaft 工具预览)
开始规划迁移
3.6
KRaft 功能继续完善,支持 ZK→KRaft 迁移
本文示例基线,不是 ZK 模式终版
3.7
ZK 模式标记 Deprecated
强制规划迁移
3.8
-
-
3.9
ZK 模式最后维护版本
迁移窗口期
选型建议 :ZooKeeper 模式只用于存量集群维护。新集群使用 Kafka 4.x/KRaft;存量 3.x 集群应先升级到受迁移路径支持的版本,再按官方流程迁移。
第 2 章 部署规划与系统调优
本章是”地基”。Kafka 性能问题 80% 源于部署前的规划不当和 OS 参数未调优。
2.1 集群规模规划 2.1.1 节点数量
集群规模
Broker 数
ZooKeeper 数
适用场景
测试
1~3
1(可内嵌)
开发/测试
小型生产
3
3
日吞吐 < 1TB,可容忍单节点故障
中型生产
5~7
3 或 5
日吞吐 1~10TB
大型生产
9~20+
5
日吞吐 > 10TB,核心业务
超大型
30+
5(ZK 会成为瓶颈)
建议迁 KRaft
经验法则 :
Broker 数 ≥ max(replication.factor, 3),且建议为奇数(便于分布均衡)
ZooKeeper 集群 3 节点可容忍 1 台故障,5 节点可容忍 2 台;超过 7 台反而降低性能 (写入需多数派确认)
2.1.2 单节点规格
组件
CPU
内存
磁盘
网络
备注
Broker(通用)
16~32 核
32~64 GB
SSD/NVMe,RAID10
10~25 Gbps
内存一半给 PageCache
Broker(高吞吐)
32~64 核
128 GB
NVMe × 多块,JBOD
25 Gbps+
多目录分摊
ZooKeeper
4~8 核
8~16 GB
SSD(事务日志必须 SSD)
10 Gbps
内存按 max(2GB, heap)
监控节点
8 核
16 GB
200 GB SSD
1 Gbps
Prometheus + Grafana
内存分配原则(Broker) :
1 2 3 JVM Heap = 6 GB ~ 12 GB(不要超过 12 GB,否则 GC 停顿不可控) 剩余内存全部留给 OS PageCache(Kafka 主要靠 PageCache 缓存热数据) 即:32 GB 内存 → Heap 8 GB + PageCache ~20 GB
⚠️ 常见误区 :给 Kafka 分配 64GB Heap。后果是 Full GC 停顿数十秒,Broker 假死,触发大量 Rebalance。Kafka 的堆不该大 。
2.1.3 磁盘规划 磁盘容量估算公式 :
1 2 3 单分区所需空间 = 平均消息大小 × 消息速率(条/s) × 保留时长(s) × 副本数 × 压缩比系数 压缩比系数:无压缩= 1 ,gzip≈0.3 ,snappy≈0.5 ,lz4≈0.5 ,zstd≈0.3 建议预留 30 % 余量
示例计算 :
1 2 3 4 业务:订单消息,平均 1KB,峰值 50000 条/s,保留 3 天,副本 3,lz4 压缩 单分区组空间 = 1KB × 50000 × 259200s × 3 × 0.5 ≈ 19.4 TB(单副本组) 实际需 ×1.3 余量 ≈ 25 TB 分摊到 5 个 Broker → 每节点 5 TB 数据盘
磁盘数量与类型 :
方案
优点
缺点
推荐场景
RAID10 (SSD)
单盘故障不影响,运维简单
成本高,容量利用率 50%
核心集群,推荐
JBOD (多块独立盘)
容量利用率 100%,成本低
单盘故障需处理,副本重建 IO 大
大规模集群,需配合 log.dir 多目录
NVMe 单盘
极高吞吐
单点故障
超高吞吐场景
JBOD 配置示例 (server.properties):
1 2 log.dirs =/data1/kafka,/data2/kafka,/data3/kafka,/data4/kafka
⚠️ 注意事项 :如果使用 JBOD,某块盘满了会导致整个 Broker 进入 Offline 状态(Kafka 3.x 已改进为单目录失效,但仍有风险)。建议每块盘保留 15% 以上余量,并配置告警。
2.2 操作系统调优(关键!) 2.2.1 文件句柄数 Kafka 每个 Segment、每个连接都占文件句柄。必须 调大。
1 2 3 4 5 6 7 8 9 10 11 * soft nofile 1000000 * hard nofile 1000000 root soft nofile 1000000 root hard nofile 1000000 * soft nproc 65535ulimit -n
若用 systemd 管理,还需在 unit 文件加 LimitNOFILE=1000000。
2.2.2 虚拟内存 / Swap 1 2 3 4 5 6 7 swapoff -a sysctl -w vm.swappiness=1echo 'vm.swappiness=1' >> /etc/sysctl.conf
2.2.3 网络参数(高吞吐必备) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 cat >> /etc/sysctl.conf <<'EOF' net.core.somaxconn = 32768 net.core.netdev_max_backlog = 32768 net.core.wmem_default = 262144 net.core.rmem_default = 262144 net.core.wmem_max = 2097152 net.core.rmem_max = 2097152 net.ipv4.tcp_rmem = 4096 65536 16777216 net.ipv4.tcp_wmem = 4096 65536 16777216 net.ipv4.tcp_tw_reuse = 1 net.ipv4.tcp_max_tw_buckets = 262144 net.ipv4.tcp_max_syn_backlog = 32768 net.ipv4.tcp_syncookies = 1 net.ipv4.tcp_fin_timeout = 30 net.ipv4.tcp_mtu_probing = 1 EOF sysctl -p
2.2.4 文件系统与挂载 1 2 3 4 5 6 7 mkfs.xfs -f /dev/nvme0n1 mount -o noatime,nodiratime /dev/nvme0n1 /data1 /dev/nvme0n1 /data1 xfs defaults,noatime,nodiratime 0 0
Kafka 官方没有给出“XFS 必然优于 ext4”的通用结论。生产选型应关注顺序吞吐、fsync 延迟、故障恢复时间及团队运维能力。
2.2.5 时钟同步(务必!) Kafka 依赖时间戳做日志保留、副本延迟判断。时间漂移会导致:
日志被提前/延后删除
副本被判为落后,ISR 频繁抖动
监控指标错乱
1 2 3 4 5 6 7 8 9 10 11 yum install -y chronycat >> /etc/chrony.conf <<'EOF' server ntp.aliyun.com iburst server time1.cloud.tencent.com iburst EOF systemctl enable chronyd --now chronyc sources -v chronyc tracking
2.2.6 关闭透明大页(THP) THP 会导致内存分配抖动,影响 Kafka 延迟稳定性。
1 2 3 4 echo never > /sys/kernel/mm/transparent_hugepage/enabledecho never > /sys/kernel/mm/transparent_hugepage/defrag
2.2.7 JDK 选择
JDK
建议
说明
JDK 8
Kafka 3.6.2 可用但已弃用
仅用于遗留兼容,不建议新部署
JDK 11
推荐
Kafka 3.6.2 构建与测试版本
JDK 17
推荐
Kafka 3.6.2 已构建与测试,先验证 GC 与依赖兼容性
其他 JDK
按具体 Kafka 小版本核对
不跨版本套用支持结论
第 3 章 ZooKeeper 集群部署
Kafka 3.x 的元数据、Controller 选举、ACL、配额、动态配置全部依赖 ZooKeeper。ZK 不稳定 = Kafka 不稳定 。
3.1 ZooKeeper 版本选择
Kafka 3.x 官方推荐 ZooKeeper 3.5.7+ ,实测 3.6.x / 3.7.x / 3.8.x 也可用。
不要用 ZooKeeper 3.4.x(老版本,缺少动态重配置等能力)。
三节点起步,独立部署(绝不要 把 ZK 和 Kafka 混部在同一台机器上!)。
3.2 部署规划
主机
角色
myid
端口
zk-node1 (10.0.0.11)
ZK
1
2181 / 2888 / 3888
zk-node2 (10.0.0.12)
ZK
2
2181 / 2888 / 3888
zk-node3 (10.0.0.13)
ZK
3
2181 / 2888 / 3888
3.3 安装与配置 1 2 3 4 5 6 7 8 cd /opt wget https://archive.apache.org/dist/zookeeper/zookeeper-3.8.4/apache-zookeeper-3.8.4-bin.tar.gz tar -xzf apache-zookeeper-3.8.4-bin.tar.gzln -s /opt/apache-zookeeper-3.8.4-bin /opt/zkmkdir -p /data/zookeeper/data /data/zookeeper/logs /var/log/zookeeper
配置文件 /opt/zk/conf/zoo.cfg :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 tickTime =2000 initLimit =10 syncLimit =5 clientPort =2181 dataDir =/data/zookeeper/data dataLogDir =/data/zookeeper/logs maxClientCnxns =200 autopurge.snapRetainCount =3 autopurge.purgeInterval =24 server.1 =10.0.0.11:2888:3888 server.2 =10.0.0.12:2888:3888 server.3 =10.0.0.13:2888:3888 4lw.commands.whitelist =ruok,srvr,mntr,stat,conf admin.enableServer =false admin.serverPort =8080
设置 myid (每台机器不同):
1 2 3 4 5 6 echo 1 > /data/zookeeper/data/myidecho 2 > /data/zookeeper/data/myidecho 3 > /data/zookeeper/data/myid
3.4 启动与验证 1 2 3 4 5 6 7 8 9 10 11 12 13 14 /opt/zk/bin/zkServer.sh start /opt/zk/bin/zkServer.sh statusecho ruok | nc 127.0.0.1 2181 echo stat | nc 127.0.0.1 2181 echo mntr | nc 127.0.0.1 2181 echo srvr | nc 127.0.0.1 2181 /opt/zk/bin/zkCli.sh -server 10.0.0.11:2181
四字命令对照表 :
命令
作用
运维用途
ruok
服务器是否运行
存活探针
stat
状态 + 连接列表
快速看角色/连接数
srvr
服务器信息(无连接)
看 ZK 版本、延迟
mntr
机器可读指标
Prometheus 采集
cons
客户端连接详情
排查谁连上来了
dump
会话与临时节点
排查会话泄漏
conf
服务器配置
核对配置
crst
重置连接统计
-
wchs/wchc/wchp
Watch 信息
数值过大说明客户端有问题
dirs
数据目录大小
磁盘增长监控
3.5 ZooKeeper 关键监控指标 1 2 3 4 5 6 7 8 9 10 11 12 zk_server_state zk_followers zk_synced_followers zk_pending_syncs zk_avg_latency zk_max_latency zk_outstanding_requests zk_num_alive_connections zk_znode_count zk_watch_count zk_ephemerals_count zk_open_file_descriptor_count
告警阈值建议 :
zk_avg_latency > 100ms → 警告
zk_outstanding_requests > 10 持续 1 分钟 → 严重
zk_pending_syncs > 0 持续 → 严重
zk_znode_count 突增 → 排查是否有异常客户端
zk_followers != zk_synced_followers → ZK 集群同步异常
3.6 ZooKeeper 数据清理 未启用 autopurge 时,ZK 的快照和事务日志会持续增长。autopurge 会按快照保留数清理可安全删除的快照及事务日志。
1 2 /opt/zk/bin/zkCleanup.sh -n 3
禁止 分别按文件时间手工删除 snapshot.* 和 log.*。两类文件共同构成恢复链,独立删除可能留下无法恢复的组合。日常使用 autopurge,人工处置使用 zkCleanup.sh/PurgeTxnLog。
3.7 ZooKeeper systemd 服务 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 [Unit] Description =Apache ZooKeeperAfter =network.target[Service] Type =forkingUser =zookeeperGroup =zookeeperEnvironment ="JAVA_HOME=/usr/lib/jvm/java-11-openjdk" ExecStart =/opt/zk/bin/zkServer.sh startExecStop =/opt/zk/bin/zkServer.sh stopExecReload =/opt/zk/bin/zkServer.sh restartRestart =on -failureRestartSec =10 LimitNOFILE =1000000 [Install] WantedBy =multi-user.target
1 2 systemctl daemon-reload systemctl enable zookeeper --now
3.8 ZooKeeper 生产配置调优
参数
默认
建议
说明
tickTime
2000
2000
保持默认,改小会增加网络负担
initLimit
10
10~20
数据量大时需加大
syncLimit
5
5
网络差可加大
maxClientCnxns
60
按规模设置,如 200
单 IP 连接上限,不建议设为无限
autopurge.purgeInterval
0
24
自动清理
autopurge.snapRetainCount
3
3~10
保留份数
jute.maxbuffer
1MB
4MB
znode 大时需调(Kafka 配额/ACL 多时)
zookeeper.sessionTimeoutMs
Kafka 侧配置
18000
Kafka 侧参数 ,见 5.x 章
globalOutstandingLimit
1000
4000
高吞吐可加大
JVM 调优(/opt/zk/conf/zookeeper-env.sh) :
1 2 3 4 5 6 7 export JAVA_HOME=/usr/lib/jvm/java-11-openjdkexport JVMFLAGS="-Xms4g -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=100 \ -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/data/zookeeper/dump \ -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9999 \ -Dcom.sun.management.jmxremote.authenticate=false \ -Dcom.sun.management.jmxremote.ssl=false"
为什么 ZK 堆不要太大 :ZK 的瓶颈在磁盘 fsync 和网络,不在内存。大堆只会带来长 GC 停顿,而 ZK 停顿会导致整个 Kafka 集群选主异常。
第 4 章 Kafka 集群部署 4.1 部署规划
主机
角色
broker.id
端口
kafka-node1 (10.0.0.21)
Broker + Controller 候选
1
9092
kafka-node2 (10.0.0.22)
Broker + Controller 候选
2
9092
kafka-node3 (10.0.0.23)
Broker + Controller 候选
3
9092
Controller 说明 :ZK 模式下,所有 Broker 都是 Controller 候选,由 ZK 选出一个。Controller 承担分区 Leader 选举、元数据变更等职责。节点数少时负担轻,节点多(>50)时 Controller 可能成为瓶颈。
4.2 安装 1 2 3 4 5 6 7 8 9 cd /opt wget https://archive.apache.org/dist/kafka/3.6.2/kafka_2.13-3.6.2.tgz tar -xzf kafka_2.13-3.6.2.tgzln -s /opt/kafka_2.13-3.6.2 /opt/kafkamkdir -p /data1/kafka /data2/kafkamkdir -p /var/log/kafka
4.3 生产级 server.properties 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 broker.id =1 delete.topic.enable =true auto.create.topics.enable =false listeners =PLAINTEXT://0.0.0.0:9092 advertised.listeners =PLAINTEXT://10.0.0.21:9092 num.network.threads =8 num.io.threads =16 socket.send.buffer.bytes =102400 socket.receive.buffer.bytes =102400 socket.request.max.bytes =104857600 log.dirs =/data1/kafka,/data2/kafka num.partitions =6 default.replication.factor =3 auto.leader.rebalance.enable =true offsets.topic.replication.factor =3 offsets.topic.num.partitions =50 transaction.state.log.replication.factor =3 transaction.state.log.min.isr =2 transaction.state.log.num.partitions =50 log.retention.hours =168 log.retention.bytes =-1 log.segment.bytes =1073741824 log.retention.check.interval.ms =300000 log.message.timestamp.type =CreateTime log.message.timestamp.difference.max.ms =9223372036854775807 log.cleanup.policy =delete log.flush.interval.messages =9223372036854775807 min.insync.replicas =2 replica.lag.time.max.ms =30000 num.replica.fetchers =4 replica.fetch.min.bytes =1 replica.fetch.max.bytes =10485760 replica.fetch.wait.max.ms =500 unclean.leader.election.enable =false replica.fetch.response.max.bytes =10485760 zookeeper.connect =10.0.0.11:2181,10.0.0.12:2181,10.0.0.13:2181 zookeeper.session.timeout.ms =18000 zookeeper.connection.timeout.ms =18000 zookeeper.max.in.flight.requests =10 background.threads =10 group.initial.rebalance.delay.ms =3000 offsets.topic.compression.codec =0
4.4 log4j.properties 调优(避免日志打满磁盘) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 log4j.rootLogger =INFO, stdout, kafkaAppender log4j.appender.kafkaAppender =org.apache.log4j.DailyRollingFileAppender log4j.appender.kafkaAppender.DatePattern ='.'yyyy-MM-dd-HH log4j.appender.kafkaAppender.File =${kafka.logs.dir}/server.log log4j.appender.kafkaAppender.layout =org.apache.log4j.PatternLayout log4j.appender.kafkaAppender.layout.ConversionPattern =[%d] %p %m (%c)%n log4j.logger.kafka.controller =INFO, controllerAppender log4j.logger.kafka.request.logger =WARN, requestAppender log4j.logger.kafka.network.Processor =WARN, requestAppender log4j.logger.kafka.server.KafkaApis =WARN, requestAppender log4j.logger.kafka.server.ReplicaFetcherThread =ERROR log4j.logger.kafka.log.LogCleaner =INFO, cleanerAppender log4j.logger.state.change.logger =INFO, stateChangeAppender
4.5 JVM 调优(kafka-server-start.sh 或独立 env 文件) 通过 systemd 的 Environment= 或独立环境文件设置,不直接修改发行包内的 kafka-server-start.sh:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 export KAFKA_HEAP_OPTS="-Xms8g -Xmx8g" export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC \ -XX:MaxGCPauseMillis=20 \ -XX:InitiatingHeapOccupancyPercent=35 \ -XX:G1HeapRegionSize=16M \ -XX:MetaspaceSize=96m \ -XX:+ExplicitGCInvokesConcurrent \ -XX:+HeapDumpOnOutOfMemoryError \ -XX:HeapDumpPath=/data/kafka/dump \ -Djava.awt.headless=true \ -Djava.net.preferIPv4Stack=true" export JMX_PORT=9999export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \ -Dcom.sun.management.jmxremote.port=9999 \ -Dcom.sun.management.jmxremote.rmi.port=9999 \ -Djava.rmi.server.hostname=127.0.0.1 \ -Dcom.sun.management.jmxremote.authenticate=false \ -Dcom.sun.management.jmxremote.ssl=false"
GC 参数说明 :
参数
说明
-Xms8g -Xmx8g
堆固定,避免动态扩容带来停顿
-XX:+UseG1GC
大堆首选,停顿可控
-XX:MaxGCPauseMillis=20
G1 目标停顿,Kafka 官方推荐值
-XX:InitiatingHeapOccupancyPercent=35
提前触发并发标记,降低 Full GC 概率
-XX:+ExplicitGCInvokesConcurrent
把 System.gc() 转为并发 GC(避免第三方库触发 STW)
-XX:+HeapDumpOnOutOfMemoryError
OOM 时 dump,便于复盘
G1 是 Kafka 3.6.2 的稳妥基线。若使用 ZGC/Shenandoah,应先按实际流量、堆大小和延迟目标压测,不应仅凭 GC 类型下结论。
4.6 启动与验证 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.propertiestail -f /var/log/kafka/server.log jps -l | grep kafka netstat -tlnp | grep 9092 /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 10.0.0.21:9092 /opt/kafka/bin/kafka-topics.sh --bootstrap-server 10.0.0.21:9092 --list
启动日志关键词检查 :
1 2 3 4 [KafkaServer id =1] started ← 启动成功 [KafkaServer id =1] Transition from STARTING to RUNNING [Controller id =1] Controller started ← 当选 Controller Creating /brokers/ids/1 .. . ← 注册到 ZK
异常关键词 :
1 2 3 4 NOT_LEADER_FOR_PARTITION Connection to node -1 ... ← 集群连不上 Unhandled exception InconsistentBrokerIdException ← broker.id 与数据目录不一致
4.7 systemd 服务 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 [Unit] Description =Apache KafkaAfter =network.target zookeeper.serviceWants =zookeeper.service[Service] Type =simpleUser =kafkaGroup =kafkaEnvironment ="JAVA_HOME=/usr/lib/jvm/java-11-openjdk" Environment ="KAFKA_HEAP_OPTS=-Xms8g -Xmx8g" Environment ="JMX_PORT=9999" Environment ="LOG_DIR=/var/log/kafka" ExecStart =/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.propertiesExecStop =/bin/kill -TERM $MAINPID Restart =on -failureRestartSec =30 LimitNOFILE =1000000 TimeoutStopSec =180 [Install] WantedBy =multi-user.target
1 2 3 systemctl daemon-reload systemctl enable kafka --now systemctl status kafka
systemd 已跟踪主进程,使用 $MAINPID 发送 SIGTERM,避免 kafka-server-stop.sh 通过进程匹配误伤同机其他 Kafka 进程。
4.8 平滑停机流程(避免数据丢失) 1 2 3 4 5 6 7 8 9 10 11 /opt/kafka/bin/kafka-topics.sh --bootstrap-server 10.0.0.21:9092 \ --describe | grep "Leader: 1" systemctl stop kafka
TERM vs KILL :
SIGTERM:Kafka 执行受控关闭(unregister、flush、清理),必须用这个
SIGKILL:强制杀,可能导致:ISR 骤减触发选主、日志文件未正常关闭、恢复时需做 log recovery(慢)
第 5 章 服务端关键配置详解 5.1 可靠性三兄弟(必须一起设计) 1 可靠性 = f (replication.factor , min.insync .replicas , acks)
组合
可靠性
可用性
适用场景
RF=1, acks=1
无(单点)
高
测试
RF=3, min.insync=1, acks=all
中(可容忍丢数据)
高
日志类可丢数据
RF=3, min.insync=2, acks=all
高(推荐)
中(挂 2 台不可用)
核心业务、金融
RF=2, min.insync=2, acks=all
中
低(挂 1 台不可用)
资源受限
配置方法(服务端默认值) :
1 2 default.replication.factor =3 min.insync.replicas =2
Topic 级别 (推荐按业务显式设置):
1 2 3 kafka-configs.sh --bootstrap-server 10.0.0.21:9092 --alter \ --entity-type topics --entity-name order-topic \ --add-config min.insync.replicas=2,retention.ms=604800000
生产者侧 :
1 2 3 4 acks =all enable.idempotence =true # 幂等,防重复 retries =2147483647 max.in.flight.requests.per.connection =5 # 幂等下可到 5
5.2 核心参数完整说明 5.2.1 网络与线程
参数
默认
建议
说明
num.network.threads
3
8~16
处理网络收发,= CPU核数/2
num.io.threads
8
16~32
处理磁盘 IO,= 磁盘数 × 2 或 = CPU核数
queued.max.requests
500
500~1000
请求队列上限,满了会阻塞网络线程(背压)
socket.send.buffer.bytes
102400
1048576
发送缓冲,高带宽可加大
socket.receive.buffer.bytes
102400
1048576
接收缓冲
socket.request.max.bytes
104857600
104857600
单请求最大字节(100MB)
connections.max.idle.ms
600000
540000
空闲连接回收(10 分钟)
背压检测 :当 queued.max.requests 满时,Kafka 停止从 socket 读取 → 生产者感知到 TCP 窗口变小 → 发送变慢。这是保护机制 ,但持续触发说明 IO 跟不上。
5.2.2 日志与保留
参数
默认
建议
说明
log.dirs
/tmp/kafka-logs
独立数据盘
绝不要用 /tmp
num.partitions
1
按业务
新建 Topic 默认分区数
log.segment.bytes
1073741824
1GB
段大小,权衡文件数与清理粒度
log.retention.hours
168
按业务
保留时长
log.retention.bytes
-1
按容量
每分区保留上限(-1 不限)
log.retention.check.interval.ms
300000
300000
清理检查频率
log.roll.hours
168
-
段滚动时间(兜底)
log.cleanup.policy
delete
按业务
delete/compact/(compact,delete)
log.index.interval.bytes
4096
4096
索引密度,影响查找到位速度
log.index.size.max.bytes
10485760
10485760
索引文件大小
segment 大小的权衡 :
太小(如 100MB):文件数暴增,inode 压力大,文件名管理开销大
太大(如 4GB):清理不及时,空间回收粒度粗
1GB 是官方默认,通常是好选择
5.2.3 副本与 ISR
参数
默认
建议
说明
replica.lag.time.max.ms
30000
30000
核心参数 ,Follower 落后多久被踢出 ISR
num.replica.fetchers
1
4~8
副本拉取线程数,副本多时加大
replica.fetch.max.bytes
1048576
10485760
单次拉取最大字节
replica.fetch.min.bytes
1
1
单次拉取最小字节(1 表示立即返回)
replica.fetch.wait.max.ms
500
500
无数据时等待时长
replica.socket.receive.buffer.bytes
65536
1048576
副本传输缓冲
replica.socket.timeout.ms
30000
30000
副本请求超时
unclean.leader.election.enable
false
false
生产必须 false
auto.leader.rebalance.enable
true
true
自动均衡 Leader
leader.imbalance.check.interval.seconds
300
300
均衡检查间隔
leader.imbalance.per.broker.percentage
10
10
不均衡容忍阈值(%)
num.replica.fetchers 调优经验 :分区数 > 1000 或副本同步频繁告警时,从默认 1 提到 4~8。注意每个 fetcher 会占用一个到 Leader 的连接,过多会增加网络开销。
5.2.4 日志清理器(Log Cleaner)
参数
默认
说明
log.cleaner.enable
true
是否启用(compact 策略必须 true)
log.cleaner.threads
1
清理线程数,一般 1~4
log.cleaner.io.max.bytes.per.second
∞
清理限速,避免抢 IO
log.cleaner.dedupe.buffer.size
128MB
去重缓冲
log.cleaner.io.buffer.size
512KB
清理 IO 缓冲
log.cleaner.backoff.ms
15000
无脏数据时的退避
log.cleaner.min.cleanable.ratio
0.5
脏数据占比阈值
log.cleaner.delete.retention.ms
86400000
墓碑消息(删除标记)保留时长
compact 场景 :Kafka Connect 的 offset topic、CDC 场景。必须 限制 key 空间,否则内存和磁盘都会爆。
5.2.5 消费者组协调
参数
默认
建议
说明
offsets.topic.replication.factor
3
3
__consumer_offsets 副本数
offsets.topic.num.partitions
50
50
__consumer_offsets 分区数
offsets.topic.segment.bytes
104857600
104857600
-
group.initial.rebalance.delay.ms
3000
0~3000
首次 rebalance 延迟(多消费者同时启动时避免风暴)
group.min.session.timeout.ms
6000
6000
会话最小超时
group.max.session.timeout.ms
1800000
300000
会话最大超时
offsets.retention.minutes
10080(7天)
10080
无消费者组 offset 保留时长
offsets.commit.timeout.ms
5000
5000
offset 提交超时
5.2.6 配额(防单客户端打满集群) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 kafka-configs.sh --bootstrap-server 10.0.0.21:9092 --alter \ --add-config 'producer_byte_rate=10485760' \ --entity-type clients --entity-name app-order kafka-configs.sh --alter --bootstrap-server 10.0.0.21:9092 \ --add-config 'consumer_byte_rate=20971520' \ --entity-type clients --entity-name app-report kafka-configs.sh --alter --bootstrap-server 10.0.0.21:9092 \ --add-config 'request_percentage=200' \ --entity-type users --entity-name app-order kafka-configs.sh --describe --bootstrap-server 10.0.0.21:9092 \ --entity-type clients --entity-name app-order
配额类型 :
producer_byte_rate:生产者字节速率(默认客户端级 + 用户级叠加)
consumer_byte_rate:消费者字节速率
request_percentage:CPU 使用百分比上限
controller_mutation_rate:Controller 变更速率(防元数据风暴)
5.3 动态配置(不重启生效) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 kafka-configs.sh --bootstrap-server 10.0.0.21:9092 --alter \ --entity-type brokers --entity-name 1 \ --add-config 'log.retention.hours=72,num.io.threads=24' kafka-configs.sh --bootstrap-server 10.0.0.21:9092 --alter \ --entity-type topics --entity-name order-topic \ --add-config 'retention.ms=259200000,max.message.bytes=2097152' kafka-configs.sh --bootstrap-server 10.0.0.21:9092 --alter \ --entity-type clients --entity-name app-x \ --add-config 'producer_byte_rate=5242880' kafka-configs.sh --bootstrap-server 10.0.0.21:9092 --describe \ --entity-type brokers --entity-name 1 --all
支持动态修改的 Broker 参数 (部分):
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 log.retention .hours / .ms / .minutes log.retention .bytes log.segment .bytes log.cleanup .policy log.cleaner .threads min.insync .replicas max.connections max.connections .per .ip num.io .threads num.network .threads unclean.leader .election .enable message.max .bytes log.flush .interval .messages log.flush .interval .ms replica.fetch .max .bytes ...
⚠️ num.io.threads / num.network.threads 动态修改后需逐个 Broker 操作,且部分版本需重启才真正生效,建议验证。
5.4 Topic 级别可覆盖配置 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 cleanup.policy =compact compression.type =producer delete.retention.ms =86400000 file.delete.delay.ms =60000 flush.messages =9223372036854775807 flush.ms =9223372036854775807 index.interval.bytes =4096 max.message.bytes =1048576 message.timestamp.difference.max.ms =9223372036854775807 message.timestamp.type =CreateTime min.cleanable.dirty.ratio =0.5 min.insync.replicas =2 preallocate =false retention.bytes =-1 retention.ms =604800000 segment.bytes =1073741824 segment.index.bytes =10485760 segment.ms =604800000 unclean.leader.election.enable =false
5.5 参数配置决策树 1 2 3 4 5 6 7 8 9 需要极致可靠性(金融/订单)? ├── 是 → RF =3, min.insync.replicas =2, acks =all, unclean.leader.election =false │ compression =zstd, retention =7d └── 否 → 需要高吞吐(日志/埋点)? ├── 是 → RF =2, min.insync.replicas =1, acks =1, compression =lz4 │ retention =3d, 启用限流配额 └── 否 → 需要状态存储(CDC/Connect)? ├── 是 → cleanup.policy =compact, RF =3, min.insync =2 └── 否 → 默认配置
第 6 章 命令行工具实战
所有命令默认 --bootstrap-server 10.0.0.21:9092(可简写为变量)。建议在 .bashrc 中设置:
1 2 3 export BS=10.0.0.21:9092,10.0.0.22:9092,10.0.0.23:9092export KAFKA_HOME=/opt/kafkaexport PATH=$PATH :$KAFKA_HOME /bin
6.1 Topic 管理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 kafka-topics.sh --bootstrap-server $BS --list kafka-topics.sh --bootstrap-server $BS --list --exclude-internal kafka-topics.sh --bootstrap-server $BS --create \ --topic order-event \ --partitions 12 \ --replication-factor 3 \ --config retention.ms=604800000 \ --config min.insync.replicas=2 \ --config compression.type=zstd \ --config cleanup.policy=delete kafka-topics.sh --bootstrap-server $BS --describe --topic order-event kafka-topics.sh --bootstrap-server $BS --describe kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions kafka-topics.sh --bootstrap-server $BS --describe --unavailable-partitions kafka-topics.sh --bootstrap-server $BS --describe --at-min-isr-partitions kafka-topics.sh --bootstrap-server $BS --list --topic '^order-event$' kafka-topics.sh --bootstrap-server $BS --alter \ --topic order-event --partitions 24 kafka-configs.sh --bootstrap-server $BS --alter \ --entity-type topics --entity-name order-event \ --add-config retention.ms=259200000 kafka-configs.sh --bootstrap-server $BS --alter \ --entity-type topics --entity-name order-event \ --delete-config retention.ms kafka-topics.sh --bootstrap-server $BS --delete --topic order-event
--describe 输出解读 :
1 2 3 Topic : order-event TopicId: xxxxx PartitionCount: 12 ReplicationFactor: 3 Topic : order-event Partition: 0 Leader: 1 Replicas: 1 ,2 ,3 Isr: 1 ,2 ,3 Topic : order-event Partition: 1 Leader: 2 Replicas: 2 ,3 ,1 Isr: 2 ,3 ,1
字段
含义
异常判断
Leader
当前 Leader Broker ID
-1 表示无 Leader(分区不可用)
Replicas
副本所在的 Broker 列表(第一个为 Preferred Leader)
-
Isr
同步中的副本
Isr 长度 < Replicas 长度 = 副本滞后
Preferred Leader 的含义 :创建 Topic 时 Replicas 列表的第一个 Broker 就是”首选 Leader”。如果当前 Leader 不是它,说明发生过切换。可以触发 preferred 选举让 Leader 回到最均衡的位置。
6.2 生产与消费 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 kafka-console-producer.sh --bootstrap-server $BS --topic order-event kafka-console-producer.sh --bootstrap-server $BS --topic order-event \ --property parse.key=true --property key.separator=: kafka-console-producer.sh --bootstrap-server $BS --topic order-event < data.txt kafka-console-producer.sh --bootstrap-server $BS --topic order-event \ --producer-property acks=all \ --producer-property compression.type=lz4 kafka-console-consumer.sh --bootstrap-server $BS --topic order-event kafka-console-consumer.sh --bootstrap-server $BS --topic order-event --from-beginning kafka-console-consumer.sh --bootstrap-server $BS --topic order-event \ --property print.key=true --property print.offset=true --property print.partition=true kafka-console-consumer.sh --bootstrap-server $BS --topic order-event \ --group debug-group --from-beginning kafka-console-consumer.sh --bootstrap-server $BS --topic order-event \ --partition 0 --offset earliest kafka-console-consumer.sh --bootstrap-server $BS --topic order-event \ --partition 0 --offset 12345 --max-messages 10 kafka-console-consumer.sh --bootstrap-server $BS --topic order-event \ --property print.timestamp=true --property print.key=true
6.3 消费者组管理(重点) 1 2 3 4 5 kafka-consumer-groups.sh --bootstrap-server $BS --list kafka-consumer-groups.sh --bootstrap-server $BS --describe --group order-consumer-group
输出解读(非常重要) :
1 2 3 GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-IDorder -consumer-group order-event 0 1234567 1234600 33 consumer-1 -xxx /10.0.0.31 consumer-1 order -consumer-group order-event 1 2234567 2234567 0 consumer-2 -xxx /10.0.0.32 consumer-2
字段
含义
异常判断
CURRENT-OFFSET
已提交的消费位点(progress)
-
LOG-END-OFFSET
分区最新 offset
-
LAG
滞后消息数 = LEO - CURRENT
持续增长 = 消费能力不足
CONSUMER-ID
消费者实例 ID
空 = 该分区无人消费
HOST
消费者所在主机
-
关键判断 :
LAG 为 0 且稳定 → 健康
LAG 持续增长 → 消费者处理能力不够,需扩容消费者实例或优化处理逻辑
CONSUMER-ID 为空但有 LAG → 消费者已下线,分区无主 (可能整个组都挂了)
某分区 LAG 远大于其他 → 分区数据倾斜 (key 设计问题)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 kafka-consumer-groups.sh --bootstrap-server $BS --describe \ --group order-consumer-group --members kafka-consumer-groups.sh --bootstrap-server $BS --describe \ --group order-consumer-group --members --verbose kafka-consumer-groups.sh --bootstrap-server $BS --describe \ --group order-consumer-group --state kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --topic order-event --reset-offsets --to-earliest --execute kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --topic order-event --reset-offsets --to-latest --execute kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --topic order-event:0 --reset-offsets --to-offset 100000 --execute kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --topic order-event --reset-offsets \ --to-datetime 2026-09-10T14:00:00.000 --execute kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --topic order-event --reset-offsets --shift-by -1000 --execute kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --topic order-event --reset-offsets --to-earliest --dry-run kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --topic order-event --reset-offsets --to-current --export > offsets.csv kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \ --reset-offsets --from-file offsets.csv --execute kafka-consumer-groups.sh --bootstrap-server $BS --delete --group order-consumer-group
⚠️ 重置 offset 前提 :消费者必须全部停止 ,否则报 Assignments can only be reset if the group 'xxx' is inactive。这是保护机制。
6.4 分区重分配(kafka-reassign-partitions.sh) 三步法 (标准流程):
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 cat > topics-to-move.json <<'EOF' { "topics" : [ {"topic" : "order-event" }, {"topic" : "user-event" } ], "version" : 1 } EOF kafka-reassign-partitions.sh --bootstrap-server $BS \ --topics-to-move-json-file topics-to-move.json \ --broker-list "1,2,3" --generate
输出示例:
1 2 3 4 5 Current partition replica assignment {"version":1 ,"partitions":[{"topic":"order-event","partition":0 ,"replicas":[1 ,2 ,3 ],"log_dirs":["any","any","any"]}]} Proposed partition reassignment configuration {"version":1 ,"partitions":[{"topic":"order-event","partition":0 ,"replicas":[2 ,3 ,1 ],"log_dirs":["any","any","any"]}]}
方式B:手工编写 (精细控制,扩容场景常用)
1 2 3 4 5 6 7 { "version" : 1 , "partitions" : [ { "topic" : "order-event" , "partition" : 0 , "replicas" : [ 4 , 5 , 6 ] , "log_dirs" : [ "any" , "any" , "any" ] } , { "topic" : "order-event" , "partition" : 1 , "replicas" : [ 5 , 6 , 4 ] , "log_dirs" : [ "any" , "any" , "any" ] } ] }
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 kafka-reassign-partitions.sh --bootstrap-server $BS \ --reassignment-json-file reassign.json \ --throttle 104857600 --execute kafka-reassign-partitions.sh --bootstrap-server $BS \ --reassignment-json-file reassign.json --verify kafka-reassign-partitions.sh --bootstrap-server $BS \ --reassignment-json-file reassign.json --verify kafka-reassign-partitions.sh --bootstrap-server $BS \ --reassignment-json-file reassign.json --cancel
输出解读 :
1 2 3 Status of partition reassignment: Reassignment of partition order-event -0 is still in progress Reassignment of partition order-event -1 completed successfully
限速原理与建议值 :
场景
throttle 建议
说明
生产集群在线迁移
50~100 MB/s
避免打满网络和磁盘
低峰期迁移
200~500 MB/s
加快速度
新集群初始化
不限(-1 或省略)
无业务影响
为什么必须限速 :不限速时,副本复制会打满网卡和磁盘 IO,导致正常生产/消费请求超时,甚至触发 ISR 收缩连锁反应。
kafka-reassign-partitions.sh --describe 必须配合 --reassignment-json-file,不支持 --topic。生成候选方案应使用前面的 --topics-to-move-json-file ... --generate,再人工审查 JSON。
6.5 副本与 Leader 管理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions kafka-topics.sh --bootstrap-server $BS --describe --unavailable-partitionscat > preferred-election.json <<'EOF' {"version" :1,"partitions" :[ {"topic" :"order-event" ,"partition" :0}, {"topic" :"order-event" ,"partition" :1} ]} EOF kafka-leader-election.sh --bootstrap-server $BS \ --election-type preferred \ --path-to-json-file preferred-election.json kafka-leader-election.sh --bootstrap-server $BS \ --election-type preferred --all-topic-partitions kafka-leader-election.sh --bootstrap-server $BS \ --election-type unclean --topic order-event --partition 0
策略 :Preferred Leader 选举通常在低峰期 执行(如凌晨 2 点),避免选举期间短暂的 Leader 切换抖动。
6.6 日志段与索引工具 1 2 3 4 5 6 7 ls -lh /data1/kafka/order-event-0/
1 2 kafka-log-dirs.sh --bootstrap-server $BS --topic-list order-event --describe
1 2 3 4 5 6 7 8 9 10 11 kafka-dump-log.sh --files /data1/kafka/order-event-0/00000000000000000000.log \ --print-data-log --deep-iteration | head -20 kafka-dump-log.sh --files /data1/kafka/order-event-0/00000000000000000000.index \ --index-sanity-check kafka-dump-log.sh --files /data1/kafka/order-event-0/*.log \ --verify-index-only
索引文件解析 :
文件
作用
.log
实际消息数据
.index
offset → 物理位置(position)索引
.timeindex
时间戳 → offset 索引
.snapshot
幂等/事务生产者的状态快照
.txnindex
事务索引
leader-epoch-checkpoint
Leader 任期记录
6.7 元数据与配置查询 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 kafka-configs.sh --bootstrap-server $BS --describe --entity-type brokers --all kafka-configs.sh --bootstrap-server $BS --describe \ --entity-type brokers --entity-name 1 --all kafka-configs.sh --bootstrap-server $BS --describe \ --entity-type topics --entity-name order-event --all kafka-broker-api-versions.sh --bootstrap-server $BS kafka-broker-api-versions.sh --bootstrap-server $BS | head -3 zookeeper-shell.sh zk1:2181/kafka get /cluster/id
6.8 性能压测工具 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 kafka-producer-perf-test.sh --topic perf-test \ --num-records 1000000 \ --record-size 1024 \ --throughput -1 \ --producer-props bootstrap.servers=$BS acks=1 batch.size=16384 linger.ms=10 compression.type=lz4 kafka-consumer-perf-test.sh --bootstrap-server $BS \ --topic perf-test \ --messages 1000000 \ --show-detailed-stats kafka-e2e-latency.sh "$BS " e2e-test 10000 all 1024
压测关键指标 :
指标
含义
判断
records/sec
吞吐(条/秒)
越高越好
MB/sec
吞吐(MB/秒)
结合磁盘/网络上限判断
avg latency
平均延迟
生产要求 < 50ms
max latency
最大延迟
尖刺可能来自 GC 或刷盘
99th latency
P99 延迟
更重要,反映长尾
第 7 章 日常运维操作手册 7.1 每日巡检清单 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 #!/bin/bash BS="10.0.0.21:9092,10.0.0.22:9092,10.0.0.23:9092" DATE=$(date +%F)echo "==================== Kafka 每日巡检 $DATE ====================" echo "----- [1] 集群 Broker 列表 -----" kafka-broker-api-versions.sh --bootstrap-server $BS 2>/dev/null | grep -c "id:" | xargs echo "Broker 数量:" echo "----- [2] Topic 总数 -----" kafka-topics.sh --bootstrap-server $BS --list | wc -l | xargs echo "Topic 数量:" echo "----- [3] 副本不足分区(⚠️ 必须为 0)-----" URP=$(kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions 2>/dev/null | wc -l)echo "副本不足分区数: $URP " [ "$URP " -gt 0 ] && kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitionsecho "----- [4] 不可用分区(⚠️ 必须为 0)-----" UAP=$(kafka-topics.sh --bootstrap-server $BS --describe --unavailable-partitions 2>/dev/null | wc -l)echo "不可用分区数: $UAP " echo "----- [5] 磁盘使用率 -----" df -h | grep -E "^/dev|kafka" echo "----- [6] 消费者组滞后 TOP 10 -----" for group in $(kafka-consumer-groups.sh --bootstrap-server $BS --list 2>/dev/null); do kafka-consumer-groups.sh --bootstrap-server $BS --describe --group $group 2>/dev/null \ | awk -v g=$group 'NR>1 && $6 ~ /^-?[0-9]+$/ {print $6, g, $2, $3}' done | sort -rn | head -10echo "----- [7] 磁盘 IO 与负载 -----" iostat -x 1 2 | tail -20uptime echo "==================== 巡检完成 ===================="
核心巡检项与阈值 :
检查项
命令
健康值
告警阈值
Broker 在线数
kafka-broker-api-versions.sh
= 规划数
少 1 台即告警
副本不足分区
--under-replicated-partitions
0
> 0
不可用分区
--unavailable-partitions
0
> 0
Controller 状态
JMX ActiveControllerCount
全集群合计 = 1
≠ 1
磁盘使用率
df -h
< 70%
> 80%
消费者组 LAG
kafka-consumer-groups.sh --describe
平稳
持续增长
ISR 收缩次数
JMX IsrShrinksPerSec
0
> 0 持续
网络丢包
sar -n EDEV
0
> 0
GC 时间
GC 日志
< 1s/分钟
P99 > 1s
7.2 Controller 相关操作 1 2 3 4 5 6 7 8 9 10 11 12 /opt/zk/bin/zkCli.sh -server 10.0.0.11:2181/kafka > get /controller {"version" :1,"brokerid" :2,"timestamp" :"..." } > ls /controller_epoch > get /controller_epoch > ls /brokers/ids [1, 2, 3] > get /brokers/ids/1 {"listener_security_protocol_map" :...,"endpoints" :["PLAINTEXT://10.0.0.21:9092" ],...}
Controller 切换的常见原因 :
原因
表现
处理
Controller 所在 Broker 停机
controller_epoch +1,短暂元数据不可写
正常,等新 Controller 就绪
ZK 会话超时
Controller 频繁切换
检查 ZK 健康、网络、GC
ZK 负载过高
zookeeper.request.timeout 报警
降低 ZK 写入频率(减少动态配置操作)
网络分区
脑裂风险
检查网络
主动触发 Controller 切换 (维护场景):
7.3 Topic 生命周期管理 7.3.1 Topic 创建规范 1 2 3 4 5 6 7 8 9 10 11 12 13 14 kafka-topics.sh --bootstrap-server $BS --create \ --topic order.event.v1 \ --partitions 12 \ --replication-factor 3 \ --config retention.ms=604800000 \ --config min.insync.replicas=2 \ --config compression.type=zstd \ --config max.message.bytes=1048576 \ --config cleanup.policy=delete
分区数规划表 :
日数据量
峰值吞吐
建议分区数
说明
< 10 GB
< 10 MB/s
3~6
小业务
10~100 GB
10~50 MB/s
6~12
常规
100 GB~1 TB
50~200 MB/s
12~48
中大型
> 1 TB
> 200 MB/s
48~200
大型,需评估
⚠️ 分区数过多的代价 :
每个分区都有文件句柄、内存缓冲、Controller 元数据
分区数 > 10000 时,Controller 选举/元数据变更会明显变慢
消费者 Rebalance 时间随分区数线性增长
单集群分区总数建议 < 200,000 (ZK 模式下 ZK 会成为瓶颈)
7.3.2 Topic 扩容分区 1 2 3 4 5 kafka-topics.sh --bootstrap-server $BS --alter --topic order.event.v1 --partitions 24
分区扩容的坑 :
若生产者按 key 分区,扩容后部分 key 的分区归属会变化 (hash(key) % N 中 N 变了)
依赖 key 有序的业务必须在扩容前停止写入、消费完存量、再扩容
分区增加后建议做一次 kafka-reassign-partitions.sh 让数据均匀
7.3.3 Topic 删除 1 2 3 4 5 6 7 8 9 10 11 12 13 14 kafka-configs.sh --bootstrap-server $BS --describe --entity-type brokers --entity-name 1 --all | grep delete.topic kafka-topics.sh --bootstrap-server $BS --delete --topic old-topic kafka-topics.sh --bootstrap-server $BS --list | grep old-topic kafka-topics.sh --bootstrap-server $BS --describe --unavailable-partitions
7.4 数据目录运维 1 2 3 4 5 6 kafka-log-dirs.sh --bootstrap-server $BS --describe --topic-list order.event.v1
offsetLag 的含义 :该副本与 Leader 的 offset 差距。正常应为 0;长期 > 0 表示副本同步有问题。
磁盘满的应急处理 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 du -sh /data1/kafka/* | sort -rh | head -20 kafka-configs.sh --bootstrap-server $BS --alter \ --entity-type topics --entity-name log-topic \ --add-config retention.ms=3600000 kafka-reassign-partitions.sh --bootstrap-server $BS \ --reassignment-json-file move.json --throttle 52428800 --execute
7.5 常见运维操作速查
操作
命令
查看集群 Topic 数
kafka-topics.sh --bootstrap-server $BS --list | wc -l
查看某 Topic 消息量
kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server $BS --topic t --time -1
查看最早/最新 offset
kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server $BS --topic t --time -2/-1
查看 Topic 总消息数
所有分区 --time -1 之和 - --time -2 之和
查看分区消息数
同上,按分区求和
修改日志级别(运行时)
kafka-configs.sh --bootstrap-server $BS --alter --entity-type broker-loggers --entity-name 1 --add-config 'kafka=DEBUG'
查看 ZK 中的 Topic 配置
zkCli.sh -server zk1:2181/kafka → get /config/topics/order.event.v1
查看 ZK 中的 ACL
get /kafka-acl/Topic/order.event.v1
GetOffsetShell 实用示例 :
1 2 3 4 5 6 7 8 9 10 kafka-run-class.sh kafka.tools.GetOffsetShell \ --bootstrap-server 10.0.0.21:9092 --topic order-event --time -1
7.6 优雅重启流程(标准 SOP) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 #!/bin/bash BROKER_ID=1 BS="10.0.0.21:9092" KAFKA_HOME=/opt/kafkaecho "=== Step 1: 检查待重启 Broker 上的 Leader 分布 ===" kafka-topics.sh --bootstrap-server $BS --describe \ | awk -v id =$BROKER_ID '$6=="Leader:" && $7==id {print $2, $4}' | wc -l \ | xargs echo "该 Broker 承载的 Leader 分区数:" echo "=== Step 2: 先做 Preferred Leader 选举(可选,减少停机抖动)===" echo "=== Step 3: 停止 Broker(优雅停机)===" systemctl stop kafkaecho "=== Step 4: 等待进程完全退出 ===" while pgrep -f "kafka.Kafka" > /dev/null; do sleep 1; done echo "进程已退出" echo "=== Step 5: 启动 Broker ===" systemctl start kafkaecho "=== Step 6: 等待启动完成 ===" sleep 30tail -20 /var/log/kafka/server.log | grep -E "started|RUNNING" echo "=== Step 7: 验证集群健康 ===" kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitionsecho "(应为空)"
重启顺序原则 :
逐个重启 ,一个完全恢复(ISR 恢复满)后再动下一个
不要同时重启 Controller 所在节点 (先确认 Controller 位置)
不要在业务高峰期重启
第 8 章 监控体系搭建
“没有监控的 Kafka 集群等于没有运维。” 本章给出完整的 Prometheus + Grafana 方案。
8.1 监控三层架构 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 ┌─────────────────────────────────────────────────────────────┐ │ 基础设施层 (OS) │ │ CPU / 内存 / 磁盘 IO / 磁盘使用率 / 网络 / 文件句柄 │ │ → node_exporter │ └─────────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────────┐ │ 中间件层 (Kafka/ ZK) │ │ JMX 指标 / Broker 状态 / Topic 状态 / 消费者 LAG │ │ → kafka_exporter (danielqsj) + JMX Exporter │ └─────────────────────────────────────────────────────────────┘ │ ┌─────────────────────────────────────────────────────────────┐ │ 业务层 │ │ 生产速率 / 消费速率 / 端到端延迟 / 消息积压 / 失败率 │ │ → 应用埋点 + 自定义指标 │ └─────────────────────────────────────────────────────────────┘
8.2 方案一:kafka_exporter(轻量,推荐) 特点 :通过 Kafka 协议拉取元数据,无需 JMX,部署简单。缺点是拿不到 JVM 和内部线程指标。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 cd /opt wget https://github.com/danielqsj/kafka_exporter/releases/download/v1.7.0/kafka_exporter-1.7.0.linux-amd64.tar.gz tar -xzf kafka_exporter-1.7.0.linux-amd64.tar.gzmv kafka_exporter-1.7.0.linux-amd64 kafka_exporter /opt/kafka_exporter/kafka_exporter \ --kafka.server=10.0.0.21:9092 \ --kafka.server=10.0.0.22:9092 \ --kafka.server=10.0.0.23:9092 \ --web.listen-address=:9308 \ --topic.filter=".*" \ --group.filter=".*" \ --sasl.enabled=false \ --verbosity=1 & curl -s http://localhost:9308/metrics | grep kafka_topic_partition_current_offset | head
systemd 服务 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 [Unit] Description =Kafka ExporterAfter =network.target kafka.service[Service] Type =simpleExecStart =/opt/kafka_exporter/kafka_exporter \ --kafka.server =10.0 .0.21 :9092 \ --kafka.server =10.0 .0.22 :9092 \ --kafka.server =10.0 .0.23 :9092 \ --web.listen-address =:9308 Restart =alwaysRestartSec =10 [Install] WantedBy =multi-user.target
kafka_exporter 核心指标 :
指标
说明
kafka_brokers
Broker 数量
kafka_topic_partitions
Topic 分区数
kafka_topic_partition_current_offset
分区最新 offset
kafka_topic_partition_oldest_offset
分区最早 offset
kafka_topic_partition_in_sync_replica
ISR 数量
kafka_topic_partition_leader_is_preferred
Leader 是否为首选(1/0)
kafka_topic_partition_replicas
副本总数
kafka_topic_partition_under_replicated_partition
是否副本不足(1/0)
kafka_consumergroup_current_offset
组当前 offset
kafka_consumergroup_lag
组 LAG(最核心指标)
kafka_consumergroup_current_offset_sum
组总 offset
kafka_consumergroup_lag_sum
组总 LAG
kafka_consumergroup_members
组成员数
kafka_exporter_build_info
版本信息
8.3 方案二:JMX Exporter(完整,深度监控) 下载 JMX Exporter :
1 2 3 4 cd /opt wget https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.20.0/jmx_prometheus_javaagent-0.20.0.jarmkdir -p /opt/kafka-exportermv jmx_prometheus_javaagent-0.20.0.jar /opt/kafka-exporter/jmx_prometheus_javaagent.jar
Kafka JMX 采集配置 /opt/kafka/config/kafka-jmx-exporter.yml :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 lowercaseOutputName: true lowercaseOutputLabelNames: true rules: - pattern: 'kafka.server<type=BrokerTopicMetrics, name=(MessagesInPerSec|BytesInPerSec|BytesOutPerSec|MessagesOutPerSec), topic=(.+)><>(\w+)' name: kafka_server_brokertopicmetrics_$1 labels: topic: "$2" type: COUNTER - pattern: 'kafka.server<type=BrokerTopicMetrics, name=(MessagesInPerSec|BytesInPerSec|BytesOutPerSec), topic=(.+)><>(Count|OneMinuteRate|FiveMinuteRate|MeanRate|FifteenMinuteRate)' name: kafka_server_brokertopicmetrics_$1_$3 labels: topic: "$2" type: GAUGE - pattern: 'kafka.server<type=ReplicaManager, name=(UnderReplicatedPartitions|PartitionCount|LeaderCount|IsrShrinksPerSec|IsrExpandsPerSec)><>Value' name: kafka_server_replicamanager_$1 type: GAUGE - pattern: 'kafka.server<type=ReplicaManager, name=(UnderReplicatedPartitions|PartitionCount|LeaderCount)><>(\w+)' name: kafka_server_replicamanager_$1_$2 type: GAUGE - pattern: 'kafka.server<type=ReplicaManager, name=IsrShrinksPerSec><>Count' name: kafka_server_replicamanager_isrshrinks_total type: COUNTER - pattern: 'kafka.server<type=ReplicaManager, name=IsrExpandsPerSec><>Count' name: kafka_server_replicamanager_isrexpands_total type: COUNTER - pattern: 'kafka.network<type=RequestMetrics, name=RequestsPerSec, request=(.+)><>(Count|OneMinuteRate|MeanRate)' name: kafka_network_requestmetrics_requests_total labels: request: "$1" type: COUNTER - pattern: 'kafka.network<type=RequestMetrics, name=TotalTimeMs, request=(.+)><>(Mean|Max|99thPercentile|999thPercentile)' name: kafka_network_requestmetrics_totaltimems_$2 labels: request: "$1" type: GAUGE - pattern: 'kafka.network<type=RequestChannel, name=(RequestQueueSize|ResponseQueueSize)><>Value' name: kafka_network_requestchannel_$1 type: GAUGE - pattern: 'kafka.controller<type=KafkaController, name=(ActiveControllerCount|OfflinePartitionsCount|LeaderElectionRateAndTimeMs)><>(Value|Count)' name: kafka_controller_kafkacontroller_$1 type: GAUGE - pattern: 'kafka.log<type=Log, name=(LogEndOffset|Size|NumLogSegments), topic=(.+), partition=(.+)><>Value' name: kafka_log_log_$1 labels: topic: "$2" partition: "$3" type: GAUGE - pattern: 'kafka.log<type=LogCleanerManager, name=(cleaner-recopy-percent|max-dirty-percent)><>Value' name: kafka_log_logcleanermanager_$1 type: GAUGE - pattern: 'kafka.coordinator.group<type=GroupMetadataManager, name=(NumGroups|NumOffsets)><>Value' name: kafka_coordinator_groupmetadatamanager_$1 type: GAUGE - pattern: 'java.lang<type=Memory><HeapMemoryUsage>(\w+)' name: jvm_memory_heap_$1 type: GAUGE - pattern: 'java.lang<type=Memory><NonHeapMemoryUsage>(\w+)' name: jvm_memory_nonheap_$1 type: GAUGE - pattern: 'java.lang<type=GarbageCollector, name=(.+)><>CollectionCount' name: jvm_gc_collection_count labels: gc: "$1" type: COUNTER - pattern: 'java.lang<type=GarbageCollector, name=(.+)><>CollectionTime' name: jvm_gc_collection_time_ms labels: gc: "$1" type: COUNTER - pattern: 'java.lang<type=OperatingSystem><>(\w+)' name: jvm_os_$1 type: GAUGE - pattern: 'java.lang<type=Threading><>(\w+)' name: jvm_threading_$1 type: GAUGE
在 Kafka 启动脚本中挂载 agent :
1 2 export KAFKA_OPTS="${KAFKA_OPTS:-} -javaagent:/opt/kafka-exporter/jmx_prometheus_javaagent.jar=7071:/opt/kafka/config/kafka-jmx-exporter.yml"
然后访问 http://<broker>:7071/metrics 即可看到 Prometheus 格式指标。
8.4 Prometheus 配置 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 global: scrape_interval: 15s evaluation_interval: 15s external_labels: cluster: 'kafka-prod' alerting: alertmanagers: - static_configs: - targets: ['localhost:9093' ]rule_files: - /etc/prometheus/rules/kafka-rules.yml scrape_configs: - job_name: 'kafka-broker-jmx' static_configs: - targets: - '10.0.0.21:7071' - '10.0.0.22:7071' - '10.0.0.23:7071' labels: env: 'prod' - job_name: 'kafka-exporter' static_configs: - targets: ['10.0.0.21:9308' ] labels: env: 'prod' - job_name: 'node' static_configs: - targets: - '10.0.0.21:9100' - '10.0.0.22:9100' - '10.0.0.23:9100' - job_name: 'zookeeper' static_configs: - targets: - '10.0.0.11:9141' - '10.0.0.12:9141' - '10.0.0.13:9141'
8.5 告警规则(可直接使用) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 groups: - name: kafka-alerts interval: 30s rules: - alert: KafkaBrokerDown expr: up{job="kafka-broker-jmx"} == 0 for: 1m labels: severity: critical annotations: summary: "Kafka Broker {{ $labels.instance }} 下线" description: "Broker {{ $labels.instance }} 已超过 1 分钟无法访问,请立即检查" - alert: KafkaBrokerCountChanged expr: kafka_brokers < 3 for: 2m labels: severity: critical annotations: summary: "Kafka 集群 Broker 数量减少(当前 {{ $value }} )" - alert: KafkaUnderReplicatedPartitions expr: kafka_server_replicamanager_underreplicatedpartitions > 0 for: 5m labels: severity: warning annotations: summary: "Broker {{ $labels.instance }} 存在副本不足分区({{ $value }} 个)" description: "持续 5 分钟有副本未同步,检查 ISR 与网络" - alert: KafkaUnderReplicatedPartitionsCritical expr: kafka_server_replicamanager_underreplicatedpartitions > 10 for: 2m labels: severity: critical annotations: summary: "Broker {{ $labels.instance }} 副本不足分区数 {{ $value }} ,超过 10 个" - alert: KafkaOfflinePartitions expr: kafka_controller_kafkacontroller_offlinepartitionscount > 0 for: 1m labels: severity: critical annotations: summary: "存在离线分区({{ $value }} 个),部分数据不可读写" - alert: KafkaActiveControllerCountAbnormal expr: sum(kafka_controller_kafkacontroller_activecontrollercount) != 1 for: 2m labels: severity: critical annotations: summary: "Controller 数量异常(当前 {{ $value }} ),应为 1" - alert: KafkaIsrShrinkRateHigh expr: rate(kafka_server_replicamanager_isrshrinks_total[5m]) > 1 for: 5m labels: severity: warning annotations: summary: "ISR 收缩频繁({{ $value }} /s),副本同步不稳定" - alert: KafkaConsumerGroupLagHigh expr: sum by (consumergroup) (kafka_consumergroup_lag) > 100000 for: 10m labels: severity: warning annotations: summary: "消费者组 {{ $labels.consumergroup }} 积压 {{ $value }} 条" description: "持续 10 分钟积压超过 10 万条,检查消费能力" - alert: KafkaConsumerGroupLagCritical expr: sum by (consumergroup) (kafka_consumergroup_lag) > 1000000 for: 5m labels: severity: critical annotations: summary: "消费者组 {{ $labels.consumergroup }} 严重积压 {{ $value }} 条" - alert: KafkaConsumerGroupLagGrowing expr: deriv(sum by (consumergroup) (kafka_consumergroup_lag)[10m:1m]) > 100 for: 15m labels: severity: warning annotations: summary: "消费者组 {{ $labels.consumergroup }} 积压持续增长" - alert: KafkaConsumerGroupNoMembers expr: kafka_consumergroup_members == 0 for: 5m labels: severity: critical annotations: summary: "消费者组 {{ $labels.consumergroup }} 无活跃成员" - alert: KafkaTopicUnderReplicated expr: kafka_topic_partition_under_replicated_partition == 1 for: 10m labels: severity: warning annotations: summary: "Topic {{ $labels.topic }} 分区 {{ $labels.partition }} 副本不足" - alert: KafkaLeaderNotPreferred expr: kafka_topic_partition_leader_is_preferred == 0 for: 30m labels: severity: info annotations: summary: "Topic {{ $labels.topic }} 分区 {{ $labels.partition }} Leader 非首选" - alert: KafkaRequestQueueHigh expr: kafka_network_requestchannel_requestqueuesize > 100 for: 5m labels: severity: warning annotations: summary: "请求队列积压({{ $value }} ),IO 处理能力不足" - alert: KafkaRequestLatencyHigh expr: kafka_network_requestmetrics_totaltimems_99thpercentile{request="Produce"} > 500 for: 5m labels: severity: warning annotations: summary: "Produce 请求 P99 延迟 {{ $value }} ms,超过 500ms" - alert: KafkaNetworkIdlePercentLow expr: kafka_server_socketservermetrics_network_io_rate < 1000000 for: 5m labels: severity: warning annotations: summary: "Broker {{ $labels.instance }} 网络空闲率过低" - alert: KafkaJvmHeapUsageHigh expr: jvm_memory_heap_used / jvm_memory_heap_max > 0.85 for: 10m labels: severity: warning annotations: summary: "Broker {{ $labels.instance }} 堆内存使用率 {{ $value | humanizePercentage }} " - alert: KafkaJvmFullGcHigh expr: rate(jvm_gc_collection_time_ms{gc=~".*Old.*|.*MarkSweep.*"}[5m]) > 1000 for: 5m labels: severity: warning annotations: summary: "Full GC 耗时 {{ $value }} ms/s,停顿过长" - alert: KafkaJvmGcPauseTooLong expr: increase(jvm_gc_collection_time_ms[5m]) > 5000 for: 5m labels: severity: warning annotations: summary: "5 分钟内 GC 总耗时超过 5 秒" - name: zookeeper-alerts rules: - alert: ZooKeeperDown expr: up{job="zookeeper"} == 0 for: 1m labels: severity: critical annotations: summary: "ZooKeeper {{ $labels.instance }} 下线" - alert: ZooKeeperLatencyHigh expr: zk_avg_latency > 100 for: 5m labels: severity: warning annotations: summary: "ZooKeeper 平均延迟 {{ $value }} ms" - alert: ZooKeeperOutstandingRequests expr: zk_outstanding_requests > 10 for: 5m labels: severity: warning annotations: summary: "ZooKeeper 排队请求 {{ $value }} ,处理能力不足" - alert: ZooKeeperPendingSyncs expr: zk_pending_syncs > 0 for: 2m labels: severity: critical annotations: summary: "ZooKeeper 存在待同步请求,集群可能不一致" - alert: ZooKeeperNoLeader expr: sum(zk_server_state{state="leader"}) != 1 for: 1m labels: severity: critical annotations: summary: "ZooKeeper 集群 Leader 数量异常" - alert: ZooKeeperFileDescriptorHigh expr: zk_open_file_descriptor_count > 50000 for: 5m labels: severity: warning annotations: summary: "ZooKeeper 打开文件数过高 {{ $value }} " - name: kafka-infra-alerts rules: - alert: KafkaDiskUsageHigh expr: (1 - node_filesystem_avail_bytes{mountpoint=~"/data.*"} / node_filesystem_size_bytes{mountpoint=~"/data.*"}) > 0.8 for: 5m labels: severity: warning annotations: summary: "{{ $labels.instance }} 磁盘 {{ $labels.mountpoint }} 使用率 {{ $value | humanizePercentage }} " - alert: KafkaDiskUsageCritical expr: (1 - node_filesystem_avail_bytes{mountpoint=~"/data.*"} / node_filesystem_size_bytes{mountpoint=~"/data.*"}) > 0.9 for: 2m labels: severity: critical annotations: summary: "{{ $labels.instance }} 磁盘 {{ $labels.mountpoint }} 使用率超过 90%,即将写满!" - alert: KafkaDiskIoSaturated expr: rate(node_disk_io_time_seconds_total[5m]) > 0.9 for: 10m labels: severity: warning annotations: summary: "{{ $labels.instance }} 磁盘 {{ $labels.device }} IO 饱和度 {{ $value }} " - alert: KafkaFileDescriptorHigh expr: node_filefd_allocated / node_filefd_maximum > 0.8 for: 5m labels: severity: warning annotations: summary: "{{ $labels.instance }} 文件句柄使用率 {{ $value | humanizePercentage }} "
8.6 Grafana 关键看板 推荐现成看板 :
Kafka Exporter Overview (ID: 7589)—— 最常用的 Kafka 总览
Kafka Exporter (ID: 18763)
Kafka Lag Exporter (ID: 11264)
JVM (Micrometer) (ID: 4701)—— 看 JVM/GC
Node Exporter Full (ID: 1860)—— 主机指标
ZooKeeper (ID: 10483)
必看核心面板(自建建议) :
面板
指标
用途
Broker 存活
up{job="kafka-broker-jmx"}
一眼看节点状态
消息吞吐(入/出)
BytesInPerSec, BytesOutPerSec
容量趋势
消息速率(条/s)
MessagesInPerSec
业务量变化
副本不足分区
UnderReplicatedPartitions
最重要健康指标
ISR 收缩速率
rate(IsrShrinks_total[5m])
副本稳定性
Produce 延迟 P99
TotalTimeMs{request="Produce"}
写入性能
Fetch 延迟 P99
TotalTimeMs{request="FetchConsumer"}
读取性能
请求队列深度
RequestQueueSize
IO 瓶颈预警
消费者 LAG TOP N
topk(10, kafka_consumergroup_lag)
业务健康
Controller 数量
ActiveControllerCount
元数据稳定性
GC 时间
jvm_gc_collection_time_ms
JVM 健康
磁盘使用率
node_filesystem_*
容量预警
8.7 ZooKeeper 监控(Prometheus 方案) ZooKeeper 没有原生 Prometheus 端点,用 exporter:
1 2 3 4 5 6 wget https://github.com/carlpett/zookeeper_exporter/releases/download/v1.0.2/zookeeper_exporter-1.0.2.linux-amd64.tar.gz tar -xzf zookeeper_exporter-1.0.2.linux-amd64.tar.gz ./zookeeper_exporter --zookeeper 10.0.0.11:2181 --web.listen-address=:9141
或使用 zk-mntr 脚本 + textfile collector。核心指标见 3.5 节。
8.8 日志采集(可选) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 cat > /etc/filebeat/filebeat.yml <<'EOF' filebeat.inputs: - type : log enabled: true paths: - /var/log/kafka/*.log - /var/log/zookeeper/*.log fields: service: kafka multiline.pattern: '^\[' multiline.negate: true multiline.match: after output.elasticsearch: hosts: ["http://10.0.0.100:9200" ] index: "kafka-logs-%{+yyyy.MM.dd}" EOF
第 9 章 性能调优实战 9.1 性能调优方法论 1 2 3 4 5 6 7 8 9 10 先测量,再优化,后验证 ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ 1 . 建立基线 │──►│ 2 . 定位瓶颈 │──►│ 3 . 单点优化 │ │ 压测得出上限 │ │ 看资源&指标 │ │ 一次改一项 │ └──────────────┘ └──────────────┘ └──────────────┘ │ ┌──────────────┐ ┌───────▼──────┐ │ 5 . 固化配置 │◄──│ 4 . 回归验证 │ │ 写入模板 │ │ 对比提升幅度 │ └──────────────┘ └──────────────┘
9.2 瓶颈定位速查表
症状
可能瓶颈
排查命令
生产延迟高,磁盘 util 100%
磁盘 IO
iostat -x 1,看 %util await
生产延迟高,磁盘 util 低
网络 / 线程 / 客户端批量
sar -n DEV 1、JMX 请求队列
消费延迟高,Broker 指标正常
消费者处理能力
看消费端 CPU、业务日志
ISR 频繁收缩
网络 / 副本线程 / GC
JMX IsrShrinks、GC 日志
请求队列堆积
IO 线程不足
RequestQueueSize > 100
CPU 高但吞吐低
压缩 / 加密 / 分区过多
看 compression.type、分区数
网络打满
副本复制 / 消费者拉取
加限速、加网卡
9.3 生产者调优(服务端视角的客户端参数) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 batch.size =1048576 # 批量大小 1MB(默认 16KB) linger.ms =100 # 等待 100ms 凑批(默认 0) compression.type =lz4 # 低 CPU、压缩率适中 buffer.memory =67108864 # 发送缓冲 64MB max.in.flight.requests.per.connection =5 acks =1 # 吞吐换可靠性(谨慎) linger.ms =0 batch.size =16384 compression.type =none acks =1 acks =all enable.idempotence =true retries =2147483647 max.in.flight.requests.per.connection =5 delivery.timeout.ms =120000
压缩算法对比 :
算法
压缩率
CPU 开销
吞吐
推荐场景
none
1.0
无
最高
内网、低延迟
lz4
~2x
低
高
通用首选
snappy
~2x
低
高
与 lz4 接近
gzip
~3x
高
低
带宽受限
zstd
~3x
中
中高
存储敏感场景首选
注意 :压缩在生产者端 做,Broker 存储压缩后的数据,消费者端解压。所以压缩节省的是网络带宽 + 磁盘空间 ,代价是客户端 CPU。
9.4 消费者调优 1 2 3 4 5 6 7 8 9 10 fetch.max.bytes =57671680 # 55MB max.partition.fetch.bytes =10485760 # 10MB fetch.min.bytes =1 fetch.max.wait.ms =500 max.poll.records =500 # 单次 poll 返回条数 max.poll.interval.ms =300000 # 处理超时(重要!) session.timeout.ms =45000 # 会话超时 heartbeat.interval.ms =3000 # 心跳间隔;应显著小于 session.timeout.ms
max.poll.interval.ms 是最容易踩坑的参数 :
1 消费者处理一批消息耗时 > max.poll.interval.ms → 被踢出组 → Rebalance → 重复消费
处理逻辑慢的业务(如调用外部 API)必须加大此值(如 10 分钟 = 600000)。
9.5 Broker 端调优 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 num.io.threads =32 # = CPU 核数(IO 密集) num.network.threads =16 # = CPU 核数 / 2 num.replica.fetchers =8 # 分区多时加大 replica.fetch.max.bytes =10485760 replica.socket.receive.buffer.bytes =1048576 log.segment.bytes =1073741824 # 1GB log.index.interval.bytes =4096 # 保持默认 num.recovery.threads.per.data.dir =4 # 恢复线程(重启加速) socket.send.buffer.bytes =1048576 socket.receive.buffer.bytes =1048576 socket.request.max.bytes =104857600 log.flush.interval.messages =9223372036854775807 # 不主动刷,靠 OS log.flush.scheduler.interval.ms =9223372036854775807 log.preallocate =false
9.6 JVM 调优深入 GC 日志开启(必须!否则出问题没法查) :
1 2 3 4 5 6 7 8 9 10 11 export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC \ -Xlog:gc*:file=/var/log/kafka/gc.log:time,uptime,level,tags:filecount=10,filesize=50m \ -XX:MaxGCPauseMillis=20 \ -XX:InitiatingHeapOccupancyPercent=35" export KAFKA_JVM_PERFORMANCE_OPTS="-XX:+UseG1GC \ -Xloggc:/var/log/kafka/gc.log -XX:+PrintGCDetails \ -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime \ -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=50M"
GC 日志分析要点 :
1 2 3 4 5 6 7 grep -E "Pause Young|Pause Full" /var/log/kafka/gc.log | \ awk -F'[()]' '{print $2}' | sort | uniq -c grep -oP "Pause \w+.*?(\d+\.\d+)ms" /var/log/kafka/gc.log | \ awk '{print $NF, $0}' | sort -rn | head -10
G1 关键参数调优 :
参数
说明
调优方向
-XX:MaxGCPauseMillis=20
目标停顿
停顿超标→减小;吞吐下降→加大
-XX:InitiatingHeapOccupancyPercent=35
并发标记触发阈值
频繁 Full GC→降低;CPU 高→提高
-XX:G1HeapRegionSize=16M
Region 大小
大对象多→加大
-XX:G1NewSizePercent=5
新生代最小占比
-
-XX:G1MaxNewSizePercent=60
新生代最大占比
-
-XX:ParallelGCThreads=N
并行 GC 线程
= CPU 核数(≤8 时)
-XX:ConcGCThreads=N
并发标记线程
= ParallelGCThreads / 4
Full GC 的常见原因 :
堆太小(< 6GB 且分区数多)
元空间泄漏(大量动态类生成,如频繁创建 admin client)
第三方库调用 System.gc()(加 -XX:+ExplicitGCInvokesConcurrent 缓解)
内存泄漏(通常是应用而非 Kafka 本身)
9.7 OS 层调优回顾(关键项) 1 2 3 4 5 6 7 8 9 10 11 12 free -g echo none > /sys/block/nvme0n1/queue/scheduler blockdev --setra 4096 /dev/nvme0n1
9.8 调优案例:吞吐从 200MB/s 提升到 1.2GB/s 初始状态 :
项
值
Broker
3 台,16C32G,SATA SSD
吞吐
200 MB/s
磁盘 util
95%
Produce P99
800ms
优化过程 :
步骤
操作
效果
1
确认瓶颈:iostat 显示 %util=95%,await=20ms
确认磁盘瓶颈
2
关闭 log.preallocate(SATA SSD 上预分配产生大量写放大)
吞吐 → 320 MB/s
3
调整 num.io.threads 8→32
吞吐 → 400 MB/s
4
开启生产者 compression.type=lz4,batch.size=1MB,linger.ms=50
吞吐 → 900 MB/s(数据量减半)
5
磁盘换 NVMe,log.dirs 改为 2 块盘
吞吐 → 1.2 GB/s,P99 → 80ms
6
num.replica.fetchers 1→6
ISR 稳定,无收缩
结论 :Kafka 性能优化是系统性 的,单一参数调整通常只有 20~50% 提升,组合优化才能数量级突破。
9.9 分区数与吞吐的关系 1 理论吞吐 = min (单分区吞吐 × 分区数, 磁盘上限, 网络上限)
但是 :分区数增加会带来:
每分区一个文件句柄(句柄数 = 分区数 × 副本数 × 段数)
每分区一个 fetcher 任务(复制线程数有限)
Controller 元数据变大
消费者 Rebalance 时间变长(线性)
经验曲线 :
1 2 3 4 5 6 分区数 单分区吞吐 集群总吞吐 1 50 MB/s 50 MB/s 10 45 MB/s 450 MB/s 100 35 MB/s 3.5 GB/s 1000 20 MB/s 20 GB/s(但已被磁盘/网络限死) 10000 5 MB/s 50 GB/s(不可达,元数据成为瓶颈)
结论 :分区数不是越多越好,找到边际收益递减的拐点 (通常是单 Broker 1000~2000 分区)。
第 10 章 故障排查手册(重点)
本章是全手册的核心章节 。前面九章教你”怎么建、怎么调”,本章教你”出事了怎么办”。 排障能力 = 对原理的理解 × 对工具的熟练 × 对现场的组织能力。三者缺一不可。
10.1 排障总则(先建立方法论,再谈工具) 10.1.1 故障现场的三问 任何故障,先问清这三件事,再动手:
问题
为什么问
怎么问
影响面是什么?
决定止血优先级(全站不可用 vs 单个非核心 Topic)
几个业务方受影响?是全部还是部分?
什么时候开始的?
定位时间拐点,与监控曲线、变更记录对齐
精确到分钟,便于查监控拐点和变更窗口
最近改过什么?
80% 的故障由变更引起
发布、配置变更、扩容、客户端升级、流量变化
血泪经验 :Kafka 集群”无缘无故”变慢/变卡的情况极少,绝大多数是变更 (客户端改参数、Topic 加分区、新增高流量业务、Broker 重启)引起的。养成”先看变更记录”的习惯,能省掉一半时间。
10.1.2 标准排障流程(黄金流程) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 ① 确认现象与影响面(5 分钟内完成) ↓ ② 止血优先,不等根因 · 业务侧:降级、限流、切换备用集群、回滚客户端配置 · 集群侧:重启异常 Broker、踢掉异常消费者、临时扩容 ↓ ③ 收集现场证据(止血动作前尽量先抓现场) · 监控曲线(时间拐点) · Broker 日志(server.log / controller.log ) · JMX 指标快照 · 网络/磁盘/线程快照(jstack、iostat 、ss) ↓ ④ 定位根因(分层排查:客户端 → 网络 → Broker → 存储 → ZK) ↓ ⑤ 修复 + 验证 ↓ ⑥ 复盘 + 预防(改监控、改参数、改流程)
10.1.3 分层排查法(关键!) Kafka 的一次请求要穿过很多层,按顺序逐层排除,比乱翻日志快得多:
层
关注对象
首选命令/工具
L1 客户端
生产者/消费者参数、线程池、GC
客户端日志、GC 日志
L2 网络
丢包、重传、RTT、DNS、防火墙
ping、mtr、ss -s、netstat -s、sar -n DEV 1
L3 Broker 线程
请求队列、IO 线程、网络线程
JMX: RequestQueueSize、RequestHandlerAvgIdlePercent
L4 存储
磁盘 util、await、文件句柄、空间
iostat -x 1、lsof | wc -l、df -h
L5 ZooKeeper
ZK 延迟、会话、节点数
四字命令 mntr、stat、ZK 日志
L6 OS/JVM
CPU、内存、GC、句柄、swap
top、vmstat 1、jstat -gcutil
顺序原则 :从最外层 开始排查。客户端问题占故障总量的一半以上,不要一上来就怀疑 Kafka 服务端。
10.1.4 现场证据收集清单(故障当下就要做) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep '(id:' /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka1:9092 \ --describe --under-replicated-partitions /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 \ --all-groups --describe > /tmp/cg-$(date +%F-%H%M).txt jstack $(pgrep -f kafka.Kafka) > /tmp/jstack-$(hostname)-$(date +%F-%H%M).txt jstat -gcutil $(pgrep -f kafka.Kafka) 1000 5 > /tmp/jstat-$(hostname).txt jmap -histo:live $(pgrep -f kafka.Kafka) | head -30 > /tmp/jmap-$(hostname).txt iostat -x 1 5 > /tmp/iostat-$(hostname).txt vmstat 1 5 > /tmp/vmstat-$(hostname).txt ss -s > /tmp/ss-$(hostname).txt sar -n DEV 1 5 > /tmp/sar-$(hostname).txt tar czf /tmp/kafka-logs-$(hostname)-$(date +%F-%H%M).tgz \ /opt/kafka/logs/*.log /opt/kafka/logs/*.out
注意 :jstack / jmap 会引发 STW(Stop The World),生产环境执行前评估影响,jmap -histo:live 尤其重,尽量在低峰或已不可用时执行。
10.2 故障速查表(症状 → 原因 → 处置)
这张表覆盖了 90% 的日常故障,先查表,再深入。
症状
最可能的 3 个原因
立即处置
生产者 TimeoutException: Expiring N records
①Broker 磁盘/网络饱和 ②acks=all 但 ISR 不足 ③客户端 linger.ms/batch.size 过大
看 Broker 磁盘 util,看 ISR;临时放宽客户端超时
生产者 NotEnoughReplicasException
ISR 副本数 < min.insync.replicas
恢复掉线 Broker;临时下调 min.insync.replicas(有丢数据风险)
生产者 RecordTooLargeException
序列化后的记录/Record Batch 超过生产者请求或 Broker/Topic 批次上限
联动核对生产者 max.request.size、Broker message.max.bytes 与 Topic max.message.bytes
消费者卡住不消费
①Rebalance 风暴 ②消费者线程阻塞/GC ③分区无数据
看 kafka-consumer-groups --describe 的 LAG 与 Consumer ID
消费者 CommitFailedException
处理超 max.poll.interval.ms,被踢出组
调大 max.poll.interval.ms 或减小 max.poll.records
消费者反复 Rebalance
①session.timeout.ms 过小 ②心跳线程被阻塞 ③消费者数 > 分区数
调大 session.timeout.ms,检查单次处理耗时
OffsetOutOfRangeException
数据已被清理 / offset 被重置到越界位置
用 --reset-offsets --to-earliest 重置
Topic 删除后长期仍存在
Controller 未处理删除、Broker 离线或日志目录权限异常
查看 Controller/Broker 日志与离线 Broker,修复后重新确认 Topic 是否消失;消费者不会阻止删除
UnderReplicatedPartitions > 0 持续
Broker 掉线 / 复制限速过低 / 磁盘慢
恢复 Broker,取消无用限速
OfflinePartitionsCount > 0
该分区所有副本都不可用(多为 unclean 选举被禁 + 全部副本掉线)
恢复 Broker;最后手段用 unclean.leader.election
Leader 频繁切换
Controller 不稳定 / 网络抖动 / 副本 I/O 慢
查 controller.log 与 NetworkClient 日志
Broker 进程莫名退出
OOM Killer / 磁盘满 / ZK 会话超时
dmesg -T | grep -i oom、df -h
磁盘使用率持续 100%
保留策略不合理 / 单分区倾斜 / 段未及时清理
查大分区,缩短 retention,迁移分区
整个集群不可用
ZK 集群故障 / Controller 无法选举
优先恢复 ZK
消息丢失
acks 配置不当 / unclean 选举 / 保留期过短
见 10.5.6
消息重复
生产者重试未幂等 / 消费者自动提交
开启幂等 enable.idempotence=true + 手动提交
消费延迟持续增长
消费能力 < 生产速度
加消费者实例(到分区数上限)/ 加分区 / 优化消费逻辑
10.3 Broker 进程与启动类故障 10.3.1 Broker 启动失败(8 种常见原因) 1 2 tail -200 /opt/kafka/logs/server.log
#
报错关键字
原因
解决
1
java.net.BindException: Address already in use
端口被占(9092/9093)
ss -lntp | grep 9092 找到占用进程
2
Cluster ID mismatch / InconsistentClusterIdException
meta.properties 里的 cluster.id 与 ZooKeeper /cluster/id 不一致(常见于误挂载或拷贝其他集群的数据目录)
停止 Broker,核对 ZooKeeper 与所有 log.dirs 的集群 ID,恢复正确挂载、配置或原数据目录;不要在原数据目录中盲删 meta.properties/checkpoint
3
Fatal error during startup: kafka.common.InconsistentBrokerIdException
broker.id 与 meta.properties 中不一致
统一 server.properties 与 meta.properties
4
No space left on device
磁盘满
清理/扩容磁盘
5
ZooKeeperClientTimeoutException
连不上 ZK
nc -zv zk1 2181 逐台验证;检查 zookeeper.connect
6
Failed to acquire lock on file .lock
同一 log.dirs 被两个进程使用
ps -ef | grep kafka 杀掉残留进程
7
java.lang.OutOfMemoryError
堆太小或内存被其他进程占
调整 KAFKA_HEAP_OPTS
8
NoSuchFileException: .../meta.properties(首次启动正常)
目录权限不对
chown -R kafka:kafka /data/kafka
排查模板 :
1 2 3 4 5 6 7 8 echo "== 端口 ==" ; ss -lntp | grep -E "9092|9093|2181" echo "== 磁盘 ==" ; df -h /dataecho "== 权限 ==" ; ls -ld /data/kafka; id kafkaecho "== 残留进程 ==" ; ps -ef | grep -v grep | grep kafkaecho "== ZK 连通 ==" ; for h in zk1 zk2 zk3; do nc -zv $h 2181; done echo "== 内存 ==" ; free -gecho "== 日志尾部 ==" ; tail -50 /opt/kafka/logs/server.log
10.3.2 Broker 频繁重启 / 掉线 现象 :监控上 Broker 数量周期性波动,UnderReplicatedPartitions 周期性抖动。
根因排查顺序 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 dmesg -T | grep -i "killed process" | tail -20 grep -i "oom" /var/log/messages | tail -20 systemctl status kafka journalctl -u kafka --since "1 hour ago" | grep -i -E "exit|fail|kill" grep -E "Session expired|zookeeper session" /opt/kafka/logs/server.log | tail grep -E "Full GC|Pause" /opt/kafka/logs/gc.log | awk '{print $NF}' | sort -rn | head iostat -x 1 3
典型的”假死”场景 :Broker 进程还在,但 ZK 会话超时被踢出,导致其上的分区全部触发 Leader 切换。原因是 zookeeper.session.timeout.ms(默认 18000)小于 GC 最长停顿或磁盘 hang 时间。
处置 :
加大 zookeeper.session.timeout.ms 到 18000~30000;
优化 GC 参数(见 9.6),把 Full GC 压到 1s 以内;
若磁盘存在 hang,检查磁盘健康(smartctl -a /dev/sdX)。
10.4 客户端生产 / 消费故障
生产环境里,60% 的”Kafka 有问题”,最后都定位到客户端 。请务必先看客户端日志。
10.4.1 生产者超时:TimeoutException: Expiring N record(s) 含义 :消息在 delivery.timeout.ms 的总时限内未能完成发送。该时限包含排队、linger.ms、请求等待和重试;request.timeout.ms 是单次请求等待 Broker 响应的上限,不与其简单相加。
排查路径 :
1 2 3 4 5 6 7 8 9 10 /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka1:9092 \ --describe --under-replicated-partitions
常见组合原因与处置 :
组合
现象
处置
acks=all + ISR 收缩到 1
全部超时,业务不可用
恢复掉线 Broker;绝不要用 acks=1 掩盖问题
batch.size / linger.ms 过大
单条延迟高,批量堆积后集中超时
吞吐场景可保留,低延迟场景调小
buffer.memory 耗尽
TimeoutException: Failed to allocate memory 或发送阻塞
降低积压、排查 Broker 延迟,必要时评估增大 buffer.memory
max.in.flight.requests.per.connection > 5 + 重试
乱序 + 超高延迟
幂等模式下最多 5,建议保持默认
10.4.2 生产者发送消息”卡住不动” 先分清是”卡在客户端”还是”卡在服务端” :
1 2 3 4 jstack <pid> | grep -A 20 "kafka-producer-network-thread"
最常见原因 :某个分区 Leader 不可用 ,消息被粘在 RecordAccumulator 里按分区排队,导致整个批次都发不出去。验证 :查该 Topic 是否有分区 Leader = -1(--describe 输出中 Leader: none)。
三方参数必须同时满足,缺一不可:
1 2 3 4 5 6 7 8 9 10 message.max.bytes =10485760 # Record Batch 上限;Kafka 3.6.2 默认 1048588 kafka-configs.sh --bootstrap-server kafka1:9092 --alter \ --entity-type topics --entity-name big-topic \ --add-config max.message.bytes=10485760 max.request.size =10485760 # 客户端单请求上限
注意 :同时评估 replica.fetch.max.bytes、replica.fetch.response.max.bytes、生产者 max.request.size 和消费者 max.partition.fetch.bytes。Kafka 可为首个超限 Record Batch 放宽单次 Fetch 上限以保证前进,但参数差距过大会降低复制和消费效率。
10.4.4 消费者 Rebalance 风暴(最影响业务的问题) 现象 :消费者组状态在 Stable / PreparingRebalance / CompletingRebalance 之间反复跳动,消费几乎停滞。
三大根因 :
根因
判定方法
修复
消费者心跳超时
日志有 Member ... has left the group 且频繁
调大 session.timeout.ms(默认 45s→建议 30~60s 且需 ≥ Broker group.min.session.timeout.ms)
单次处理超时被踢
日志有 Member ... was removed from the group + max.poll.interval.ms 相关
调大 max.poll.interval.ms(默认 300s)或减小 max.poll.records
消费者数量变化 (上线/下线/重启/GC)
有部署记录或 GC 曲线
滚动发布改为分批;调大 session.timeout.ms 容忍短暂 GC
关键参数关系(务必理解) :
1 2 session.timeout.ms ≥ 3 × 心跳间隔(heartbeat.interval.ms,默认 3s ) max.poll.interval.ms > 单次 poll 拉取的所有消息的总处理时间
应急止血 :把消费逻辑改为”多线程处理 + 主线程持续 poll”,或临时把 max.poll.records 从 500 降到 50。
10.4.5 CommitFailedException 1 2 3 org.apache .kafka .clients .consumer .CommitFailedException : Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.
含义 :你还在处理上一批数据,但已经被踢出组了,此时提交 offset 必然失败。
处置 :
调大 max.poll.interval.ms;
减小 max.poll.records;
消费逻辑异步化,保证 poll 循环不停顿。
10.4.6 消费积压(LAG 持续增长) 判定公式 :
1 LAG = (Topic 最新 off set 之和) - (消费组已提交 off set 之和)
排查决策树 :
1 2 3 4 5 6 7 8 LAG 增长 ├─ 生产速度是否突增? ── 是 → 查上游流量变化,考虑限流 └─ 否 → 消费端问题 ├─ 消费者实例数 < 分区数? ── 是 → 加实例(但最多等于分区数) ├─ 消费者实例数 > 分区数? ── 是 → 多余实例空转,白加 ├─ 单条处理耗时是否变长? ── 是 → 查下游依赖(DB/HTTP) ├─ 是否频繁 Rebalance? ── 是 → 见 10.4 .4 └─ 是否消费线程被 GC /日志阻塞? ── 是 → 查 GC 日志与日志级别
处置手段(按优先级) :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 kafka-topics.sh --bootstrap-server kafka1:9092 \ --alter --topic order-event --partitions 24 fetch.min.bytes=1048576 fetch.max.wait.ms=500 max.partition.fetch.bytes=1048576 max.poll.records=500 kafka-consumer-groups.sh --bootstrap-server kafka1:9092 \ --reset-offsets --to-latest --group g1 --topic t1 --execute
10.4.7 OffsetOutOfRangeException 原因 :消费者提交的 offset 已经超出该分区现存数据范围(数据被 retention 清理,或 offset 被误重置)。
1 2 3 4 5 6 7 8 9 kafka-run-class.sh kafka.tools.GetOffsetShell \ --bootstrap-server kafka1:9092 --topic order-event --time -2 kafka-run-class.sh kafka.tools.GetOffsetShell \ --bootstrap-server kafka1:9092 --topic order-event --time -1 kafka-consumer-groups.sh --bootstrap-server kafka1:9092 \ --group g1 --topic order-event --reset-offsets --to-earliest --execute
运维提醒 :这类故障往往是”消费停了太久 > 数据保留期 “导致的,属于架构问题而非参数问题。应建立 LAG 告警(见 8.5),在积压之初就介入。
10.4.8 消费”没有数据”但 LAG 为 0 可能原因:
消费者订阅的 Topic 拼写错误(会收到空分配);
生产者写入的分区与消费者分配的分区不重叠(如只订阅了某个分区);
消费组没有有效的已提交 offset,且 auto.offset.reset=latest,因此从分区末尾开始;已有有效 offset 时该参数不生效;
消息被事务标记为未提交(isolation.level=read_committed + 事务未提交)。
10.5 副本与数据一致性故障 10.5.1 ISR 频繁收缩(集群最危险的慢性病) 判定 :JMX kafka.server:type=ReplicaManager,name=IsrShrinksPerSec 与 IsrExpandsPerSec 持续 > 0,UnderReplicatedPartitions 抖动。
根本原因 :Follower 在 replica.lag.time.max.ms(默认 30s)内没能追上 Leader。
四类原因与对策 :
原因
判定
对策
网络带宽不足
sar -n DEV 1 网卡接近线速
增加网卡/降副本同步流量(开压缩)/限速迁移
磁盘 I/O 慢
iostat -x 的 await > 20ms,%util > 90%
换 SSD、减少单盘分区、关闭 log.preallocate
复制线程不够
num.replica.fetchers 小、follower 排队
提高到 CPU 核数的一半
Full GC 停顿
GC 日志中 Pause > replica.lag.time.max.ms
调优 GC、加大堆
第一步永远先看这个 :
1 2 3 4 5 6 kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions
UnderReplicatedPartitions 是关键健康指标,但为 0 不代表集群整体健康;还要同时检查离线分区、请求延迟、磁盘容量、Controller、消费积压和错误日志。该值非 0 时应暂停非必要变更。
10.5.2 UnderReplicatedPartitions 长时间不恢复 排查三步 :
1 2 3 4 5 6 7 8 9 10 kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep -c '(id:' kafka-configs.sh --bootstrap-server kafka1:9092 --describe --broker 1 kafka-configs.sh --bootstrap-server kafka1:9092 --describe --topic order-event | grep -i throttl
清除限速 :
1 2 3 4 5 6 7 8 9 kafka-configs.sh --bootstrap-server kafka1:9092 --alter --broker 1 \ --delete-config leader.replication.throttled.rate kafka-configs.sh --bootstrap-server kafka1:9092 --alter --broker 1 \ --delete-config follower.replication.throttled.rate kafka-configs.sh --bootstrap-server kafka1:9092 --alter --topic order-event \ --delete-config leader.replication.throttled.replicas kafka-configs.sh --bootstrap-server kafka1:9092 --alter --topic order-event \ --delete-config follower.replication.throttled.replicas
10.5.3 OfflinePartitionsCount > 0(分区彻底不可用) 含义 :该分区当前没有可用 Leader。常见原因是 ISR 中没有可选副本,也可能是 Controller、元数据或 Broker 状态异常;不能仅凭该指标断言所有磁盘副本都已损坏。
处置流程 :
1 2 3 4 5 6 ① 确认涉及哪些 Broker kafka-topics.sh --describe --unavailable-partitions --bootstrap-server kafka1:9092 ② 优先恢复这些 Broker(这是根治) ③ 若 Broker 数据已损毁、无法恢复,且业务无法等待: - 确认是否存在可用的旧副本(在其他盘的副本) - 最后手段:开启 unclean 选举
开启 unclean 选举(⚠️ 会丢数据,需业务书面确认) :
1 2 unclean.leader.election.enable =true
1 2 3 4 5 6 kafka-configs.sh --bootstrap-server kafka1:9092 --alter --entity-type brokers \ --entity-default --add-config unclean.leader.election.enable=true kafka-configs.sh --bootstrap-server kafka1:9092 --alter --entity-type topics \ --entity-name order-event --add-config unclean.leader.election.enable=true
恢复后立即删除临时动态覆盖:默认 Broker 配置使用 --entity-default --delete-config unclean.leader.election.enable;Topic 覆盖使用对应 Topic 的 --delete-config。
10.5.4 Leader 频繁切换 判定 :JMX kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs 的 Count 持续增长。
可能原因 :
Controller 本身不稳定(频繁重选举);
副本所在 Broker 的 ZK 会话超时(假死);
Leader 副本所在磁盘 I/O hang,导致其被 Controller 判定为不可用。
排查 :
1 2 3 4 5 6 /opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controller grep -E "Broker.*failed|Shrinking ISR|Leader election" /opt/kafka/logs/controller.log | tail -50
10.5.5 副本同步严重落后(ReplicaLag 持续增长) JMX 指标 :kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica
常见原因 :Follower 的 fetch 速度 < Leader 的写入速度(磁盘/网络是瓶颈),或者 Follower 的 num.replica.fetchers 设置过小。
调优 :
1 2 3 4 num.replica.fetchers =6 # 默认 1,副本多时必调 replica.fetch.min.bytes =1 # 默认 1,别调大 replica.fetch.max.bytes =10485760 # 加大 replica.fetch.wait.max.ms =500 # 减小等待,加快同步
10.5.6 数据丢失的判定与根因 先建立一个认知 :Kafka 不会 因为”Broker 挂了”就丢数据——只要副本机制配置正确。丢数据几乎都由下面 5 种配置/操作导致:
#
场景
配置陷阱
预防
1
acks=1 或不设置
生产者不等副本确认
生产环境强制 acks=all
2
min.insync.replicas=1
只要 1 个副本写入就返回成功
设为 2(副本 3 时)
3
unclean.leader.election.enable=true
落后的副本被选为 Leader,截断数据
保持 false
4
retention 过短
数据被清理而消费未跟上
按最慢消费者设计保留期
5
单副本 Topic
无容灾,磁盘损毁即丢
强制 replication.factor >= 2,推荐 3
丢数据后的验证方法 :
1 2 3 4 kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server kafka1:9092 \ --topic order-event --time -1
根本解法(生产标准配置组合) :
1 2 3 4 5 6 7 8 9 default.replication.factor =3 min.insync.replicas =2 unclean.leader.election.enable =false acks =all enable.idempotence =true retries =2147483647 max.in.flight.requests.per.connection =5 # 幂等模式下安全
这四项需要组合设计,但它们不是数学乘法,也不能替代端到端校验、跨故障域部署、备份/重放能力和生产者错误处理。
10.5.7 消息重复 原因 :生产者重试 + 无幂等(网络已写入但客户端认为失败,重试导致重复)。
解决 :enable.idempotence=true(自动开启幂等 + acks=all + retries>1 + max.in.flight<=5)。
若业务无法改生产者 :消费端做幂等(唯一键去重表)。
10.6 Controller / ZooKeeper 故障 10.6.1 Controller 频繁切换 Controller 是 Kafka 集群的”大脑”(负责分区 Leader 选举、副本分配)。它由 ZK 在 Broker 中选出,正常应长期稳定。
排查 :
1 2 3 4 5 6 7 8 9 /opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controllertail -200 /opt/kafka/logs/controller.log grep -E "Controller moved|New controller|resigning|Expired|session" \ /opt/kafka/logs/controller.log /opt/kafka/logs/server.log | tail -30
三大原因 :
原因
日志特征
处置
ZK 会话超时
Session expired / Expired
加大 zookeeper.session.timeout.ms,排查 GC
Controller 所在 Broker 网络不稳
Connection to ... timed out
排查网络/交换机
ZK 集群自身不稳定
ZK 侧日志 Session 0x... timed out
见第 15 章
运维经验 :Controller 切换本身只需几百毫秒到几秒,期间的元数据变更(建 Topic、改配置)会失败 ,但数据读写不受影响 (分区 Leader 不变)。所以”Controller 切了一下”通常不是紧急故障,但频繁切换 必须查。
10.6.2 ZK 会话超时导致 Broker”假死” 见 10.3.2。核心是让 GC 停顿 < zookeeper.session.timeout.ms。
10.7 性能类故障 10.7.1 生产延迟高(Produce P99 飙高) 定位脚本 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 #!/bin/bash echo "===== 1. 磁盘 I/O =====" iostat -x 1 3 | awk '$1 !~ /^(Linux|Device|loop|$) {print}' echo "===== 2. CPU =====" top -bn1 | head -12echo "===== 3. 网络 =====" sar -n DEV 1 3 | grep -E "Average|IFACE" echo "===== 4. 内存/swap =====" free -mecho "===== 5. JVM GC =====" jstat -gcutil $(pgrep -f kafka.Kafka) 1000 3echo "===== 6. 线程数/句柄 =====" echo "threads: $(ls /proc/$(pgrep -f kafka.Kafka) /task | wc -l)" echo "fds: $(ls /proc/$(pgrep -f kafka.Kafka) /fd | wc -l)"
常见结论对照 :
现象
瓶颈
措施
%util > 90 且 await > 20ms
磁盘
换 SSD / 分散分区 / 关预分配
us+sy > 80% 且 Broker 进程占满
CPU
加大 num.io.threads/num.network.threads,或加机器
网卡接近线速
网络
开压缩、加网卡、升级万兆
jstat 中 FGC 持续增长
GC
见 9.6
RequestQueueSize 持续 > 100
Broker 处理能力不足
加 IO 线程或扩容
PageCache 命中率低
读放大
加大内存、减少 Broker 上其他进程
10.7.2 消费延迟高 见 10.4.6。补充一个易忽略点:消费者端的 receive.buffer.bytes 太小 会导致 TCP 窗口受限,跨机房消费必调。
10.7.3 磁盘 I/O 打满 必须区分 :是”写满”还是”队列积压”。看 iostat -x 的 aqu-sz(队列深度)与 %util。
1 2 3 4 5 6 7 8 iotop -oPals -lhS /data/kafka/order-event-*/ | head -20
10.7.4 句柄耗尽 1 2 3 ls /proc/$(pgrep -f kafka.Kafka)/fd | wc -l
解决 :调大 nofile(见 2.2),并注意 Kafka 的 ulimit -n 需要 ≥ (分区数 × 2) + 网络连接数 。
10.8 磁盘与数据目录故障 10.8.1 磁盘写满(最高频的 P0 故障) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 du -sh /data/kafka/* | sort -rh | head -20 python3 - <<'PY' import os, collections root="/data/kafka" agg=collections.Counter()for d in os.listdir(root): p=os.path.join(root,d) if not os.path.isdir(p): continue topic="-" .join (d.split("-" )[:-1]) agg[topic]+=sum (os.path.getsize(os.path.join(p,f)) for f in os.listdir(p) if os.path.isfile(os.path.join(p,f)))for t,s in agg.most_common(20): print (f"{s/1024**3:8.2f} GB {t}" ) PY
处置顺序(从安全到危险) :
缩短保留期 (安全,立即生效需等检查周期):1 2 3 4 5 kafka-configs.sh --bootstrap-server kafka1:9092 --alter --entity-type topics \ --entity-name log-topic --add-config retention.ms=86400000 kafka-configs.sh --bootstrap-server kafka1:9092 --alter --entity-type brokers \ --entity-name 1 --add-config log.retention.check.interval.ms=60000
限制单分区大小 :retention.bytes=107374182400
迁移分区到其他磁盘/节点 (见第 12 章)
临时扩容磁盘 (云盘在线扩容 + xfs_growfs)
最后手段 :手动删除最旧 segment(⚠️ 会丢数据,必须先确认 offset 已被消费)
10.8.2 磁盘只读 / 坏盘 现象 :日志出现 Input/output error、Read-only file system;kafka-log-dirs.sh 报 error 字段非空。
1 2 3 4 5 6 7 8 9 10 smartctl -a /dev/sdb | grep -E "Reallocated|Pending|Health" dmesg -T | grep -i -E "I/O error|medium error" kafka-log-dirs.sh --bootstrap-server kafka1:9092 --broker-list 1 \ --topic-list order-event --describe | python3 -m json.tool
警告 :绝不能在 Broker 运行中直接拔盘/umount。必须先把该盘上的分区迁走,否则对应分区会进入 Offline。
10.8.3 数据目录被误删 能恢复吗? 大概率不能。Kafka 的数据是”日志”,没有 Kafka 自身的备份机制。
能做的 :
如果还有其他副本存活 → 它们会自动补齐(前提是 Leader 没被删);
若所有副本都被删 → 该分区数据永久丢失,只能从上游数据源重放 ;
所以:数据目录的权限、备份策略、防误删(chattr +i 保护父目录)非常重要 。
10.9 排障工具箱(命令速查) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 jps -lvm jstack <pid> | grep -A 30 "RequestHandler" jstack <pid> | grep -c "java.lang.Thread.State" cat /proc/<pid>/status | grep -E "Threads|VmRSS" jstat -gcutil <pid> 1000 10 jstat -gc <pid> 1000 10 -Xlog:gc*:file=/opt/kafka/logs/gc.log:time ,uptime ,level,tags:filecount=10,filesize=100M jmap -histo:live <pid> | head -30 jmap -dump:live,format=b,file=/tmp/heap.hprof <pid> iostat -x 1 vmstat 1 sar -n DEV 1 / sar -n TCP,ETCP 1 ss -s / ss -antp | grep 9092 | wc -l pidstat -d 1 / pidstat -u 1 -p <pid> lsof -p <pid> | wc -l tcpdump -i eth0 -nn port 9092 -w /tmp/kafka.pcap tcpdump -i eth0 -nn 'port 9092 and tcp[tcpflags] & (tcp-rst) != 0' kafka-topics.sh --bootstrap-server kafka1:9092 \ --describe --under-replicated-partitions kafka-log-dirs.sh --describe --bootstrap-server kafka1:9092 kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --all-groups --describe kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 curl -O https://arthas.aliyun.com/arthas-boot.jar java -jar arthas-boot.jar java -jar jmxterm-1.0.4-uber.jar -l localhost:9999
10.10 排障决策树(贴墙版) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 业务反馈"Kafka 出问题了" │ ├─【生产发不出去 / 超时】 │ ├─ Broker 都在线? ── 否 → 恢复 Broker(10.3 ) │ └─ 是 │ ├─ UnderReplicatedPartitions > 0 ? ── 是 → ISR 问题(10.5 .1 ) │ ├─ OfflinePartitions > 0 ? ── 是 → 恢复副本/考虑 unclean(10.5 .3 ) │ ├─ Broker 磁盘/网络饱和? ── 是 → 性能问题(10.7 ) │ └─ Broker 正常 → 查客户端参数 acks/min.insync/max.request.size(10.4 ) │ ├─【消费慢 / 积压】 │ ├─ 频繁 Rebalance? ── 是 → 10.4 .4 │ ├─ 消费者数 ≥ 分区数? ── 否 → 加实例 │ ├─ 单条处理慢? ── 是 → 查下游依赖 │ └─ 都不像 → 查生产者是否突增(10.4 .6 ) │ ├─【集群规模异常:Broker 数量变化 / 分区不可用】 │ ├─ Broker 进程存活但被 ZK 踢? ── 是 → GC/会话超时(10.3 .2 ) │ ├─ 磁盘满/坏盘? ── 是 → 10.8 │ └─ Controller 频繁切换? ── 是 → 10.6 .1 │ └─【元数据操作失败:建 Topic/改配置报错】 ├─ Controller 是否正常? ── 见 10.6 .1 └─ ZK 是否正常? ── 见第 15 章
10.11 完整排障案例:一次线上”生产大面积超时” 时间线 :
时间
事件
10:02
监控告警:多业务反馈生产者超时率 30%
10:03
确认影响面:5 个业务,均为 acks=all 的 Topic
10:04
查 Broker 数量:3 台 → 2 台 (broker 3 掉线)
10:05
查 --under-replicated-partitions:142 个分区 ISR 从 3 缩到 2
10:06
查 broker 3 机器:dmesg 显示 OOM Killer 杀死了 kafka 进程
10:08
止血 :扩容 broker 3 内存至 64G,重启 Broker
10:15
ISR 逐步恢复(限速迁移中),但有 30 个分区仍卡住
10:18
查发现 broker 1 上残留了 2 天前的迁移限速 follower.replication.throttled.rate=1MB/s
10:20
清除限速,ISR 在 4 分钟内全部恢复,业务恢复
10:25
复盘:①OOM 原因是堆 8G 但 PageCache 被其他进程抢走;②限速未清理是历史遗留问题
经验总结 :
UnderReplicatedPartitions 是第一个要看、也是最能说明问题的指标;
残留限速 是极隐蔽的坑,迁移后务必确认清除;
OOM Killer 是最常见的 Broker “莫名死亡”原因,dmesg 是必查项;
生产环境建议给 Kafka 进程设置 OOMScoreAdjust=-500(systemd)降低被杀的优先级。
1 2 3 4 5 6 [Service] OOMScoreAdjust =-500 LimitNOFILE =1000000 Restart =on -failureRestartSec =10
第 11 章 安全加固(SASL / SSL / ACL)
默认的 Kafka 是完全不设防的 :任何人只要能连上 9092,就能读所有消息、删所有 Topic。 生产环境(尤其是等保、金融、政企)必须具备:认证(谁)+ 授权(能干啥)+ 加密(传什么) 。
11.1 安全模型总览
维度
机制
解决什么
加密
SSL/TLS
防止链路窃听、中间人
认证
SSL 双向认证 / SASL
确认”你是谁”
授权
ACL
确认”你能做什么”
审计
日志 + 自定义 Authorizer
事后追溯
SASL 机制对比(选型关键) :
机制
复杂度
是否需要外部依赖
适用场景
PLAIN
★
无(但需 SSL 保护)
内网快速落地,必须叠加 SSL
SCRAM-SHA-256/512
★★
无(凭据存 ZK)
推荐 ,无 Kerberos 负担,安全性好
GSSAPI
★★★★★
需要 KDC(Kerberos)
已有 Kerberos 的大企业
OAUTHBEARER
★★★
需要 OAuth 服务
云原生、需要 Token 短时有效
选型建议 :绝大多数团队选 SASL/SCRAM-SHA-512 + SSL + ACL ,兼顾安全与运维成本。
11.2 SSL 双向认证(加密 + 认证) 11.2.1 生成证书(CA + 各节点) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 mkdir -p /opt/kafka/ssl && cd /opt/kafka/ssl PASS=kafka123456 openssl req -new -x509 -keyout ca.key -out ca.crt -days 3650 -nodes \ -subj "/C=CN/ST=Beijing/L=Beijing/O=MyCompany/CN=Kafka-CA" for host in kafka1 kafka2 kafka3; do IP=$(getent ahostsv4 "$host " | awk 'NR==1 {print $1}' ) keytool -genkeypair -alias $host -keyalg RSA -keysize 2048 -validity 3650 \ -keystore ${host} .keystore.jks -storepass $PASS -keypass $PASS \ -dname "CN=${host} ,OU=Kafka,O=MyCompany,L=Beijing,ST=Beijing,C=CN" \ -ext SAN=DNS:${host} ,IP:${IP} keytool -certreq -alias $host -keystore ${host} .keystore.jks \ -file ${host} .csr -storepass $PASS -keypass $PASS \ -ext SAN=DNS:${host} ,IP:${IP} openssl x509 -req -CA ca.crt -CAkey ca.key -in ${host} .csr \ -out ${host} .crt -days 3650 -CAcreateserial \ -extfile <(printf 'subjectAltName=DNS:%s,IP:%s\n' "$host " "$IP " ) keytool -importcert -alias CARoot -file ca.crt -keystore ${host} .keystore.jks \ -storepass $PASS -keypass $PASS -noprompt keytool -importcert -alias $host -file ${host} .crt -keystore ${host} .keystore.jks \ -storepass $PASS -keypass $PASS -noprompt keytool -importcert -alias CARoot -file ca.crt \ -keystore ${host} .truststore.jks -storepass $PASS -nopromptdone openssl verify -CAfile ca.crt kafka1.crt keytool -list -v -keystore kafka1.keystore.jks -storepass "$PASS " | \ grep -A3 'SubjectAlternativeName'
11.2.2 Broker 端 SSL 配置 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 listeners =SSL://0.0.0.0:9093,PLAINTEXT://0.0.0.0:9092 advertised.listeners =SSL://kafka1:9093,PLAINTEXT://kafka1:9092 listener.security.protocol.map =SSL:SSL,PLAINTEXT:PLAINTEXT inter.broker.listener.name =PLAINTEXT ssl.keystore.location =/opt/kafka/ssl/kafka1.keystore.jks ssl.keystore.password =kafka123456 ssl.key.password =kafka123456 ssl.truststore.location =/opt/kafka/ssl/kafka1.truststore.jks ssl.truststore.password =kafka123456 ssl.client.auth =required ssl.enabled.protocols =TLSv1.2,TLSv1.3 ssl.protocol =TLSv1.3
主机名校验由客户端的 ssl.endpoint.identification.algorithm=HTTPS 控制。客户端连接地址必须出现在证书 SAN 中;现代 TLS 校验不能只依赖 CN。
11.2.3 客户端 SSL 接入(Java) 1 2 3 4 5 6 7 8 security.protocol =SSL ssl.truststore.location =/path/client.truststore.jks ssl.truststore.password =kafka123456 ssl.keystore.location =/path/client.keystore.jks ssl.keystore.password =kafka123456 ssl.key.password =kafka123456 ssl.endpoint.identification.algorithm =HTTPS
命令行工具:
1 2 3 kafka-topics.sh --bootstrap-server kafka1:9093 \ --command-config /opt/kafka/config/ssl-client.properties \ --list
11.3 SASL/SCRAM(推荐方案) 11.3.1 创建 SCRAM 用户 1 2 3 4 5 6 7 8 9 10 11 12 13 kafka-configs.sh --bootstrap-server kafka1:9092 \ --alter --add-config 'SCRAM-SHA-256=[password=admin123],SCRAM-SHA-512=[password=admin123]' \ --entity-type users --entity-name admin kafka-configs.sh --bootstrap-server kafka1:9092 \ --alter --add-config 'SCRAM-SHA-512=[password=Order@2024]' \ --entity-type users --entity-name order-app kafka-configs.sh --bootstrap-server kafka1:9092 --describe \ --entity-type users --entity-name admin
Kafka 3.6.2 已支持通过 --bootstrap-server 增删改查 SCRAM 凭据。--zookeeper 路径已弃用,只应作为 Broker 全部不可用时的受控恢复手段。
11.3.2 Broker 端 SASL_SSL 配置 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 listeners =SASL_SSL://kafka1:9094 advertised.listeners =SASL_SSL://kafka1:9094 security.inter.broker.protocol =SASL_SSL sasl.enabled.mechanisms =SCRAM-SHA-512 sasl.mechanism.inter.broker.protocol =SCRAM-SHA-512 authorizer.class.name =kafka.security.authorizer.AclAuthorizer super.users =User:admin;User:kafka allow.everyone.if.no.acl.found =false ssl.keystore.location =/opt/kafka/ssl/kafka1.keystore.jks ssl.keystore.password =kafka123456 ssl.key.password =kafka123456 ssl.truststore.location =/opt/kafka/ssl/kafka1.truststore.jks ssl.truststore.password =kafka123456 ssl.client.auth =none
11.3.3 Broker JAAS 配置(Broker 之间互相认证) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 cat > /opt/kafka/config/kafka_server_jaas.conf <<'EOF' KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin123" ; }; // 若 ZK 也开启了 SASL(生产建议),需要 Client 段 Client { org.apache.zookeeper.server.auth.DigestLoginModule required username="admin" password="admin123" ; }; EOF
1 2 export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
11.3.4 客户端接入(SASL_SSL + SCRAM) 1 2 3 4 5 6 7 8 9 10 security.protocol =SASL_SSL sasl.mechanism =SCRAM-SHA-512 sasl.jaas.config =org.apache.kafka.common.security.scram.ScramLoginModule required \ username="order-app" password="Order@2024"; ssl.truststore.location =/path/client.truststore.jks ssl.truststore.password =kafka123456
Java 代码方式:
1 2 3 4 5 props.put("security.protocol" , "SASL_SSL" ); props.put("sasl.mechanism" , "SCRAM-SHA-512" );String jaas = "org.apache.kafka.common.security.scram.ScramLoginModule required " + "username=\"order-app\" password=\"Order@2024\";" ; props.put("sasl.jaas.config" , jaas);
Python 客户端 :
1 2 3 4 5 6 7 8 producer = KafkaProducer( bootstrap_servers=['kafka1:9094' ], security_protocol='SASL_SSL' , sasl_mechanism='SCRAM-SHA-512' , sasl_plain_username='order-app' , sasl_plain_password='Order@2024' , ssl_cafile='/opt/kafka/ssl/ca.crt' , )
11.4 ACL 授权实战 11.4.1 ACL 语法 1 2 3 --add --allow-principal User:<用户名> \ --operation <操作> \ --[topic|group |cluster |transactional-id] <资源名>
资源类型
可用操作
topic
Read, Write, Create, Delete, Describe, DescribeConfigs, Alter, AlterConfigs, All
group
Read, Describe, Delete, All
cluster
Create, ClusterAction, Describe, DescribeConfigs, Alter, AlterConfigs, IdempotentWrite, All
transactional-id
Describe, Write, All
11.4.2 常用授权模板 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 BS="--bootstrap-server kafka1:9094" AUTH="--command-config /opt/kafka/config/admin-sasl.properties" kafka-acls.sh $BS $AUTH --add \ --allow-principal User:order-app \ --operation Write --operation Describe \ --topic order-event kafka-acls.sh $BS $AUTH --add \ --allow-principal User:order-consumer \ --operation Read --operation Describe \ --topic order-event kafka-acls.sh $BS $AUTH --add \ --allow-principal User:order-consumer \ --operation Read \ --group order-group kafka-acls.sh $BS $AUTH --add \ --allow-principal User:order-app \ --operation All --resource-pattern-type prefixed \ --topic order- kafka-acls.sh $BS $AUTH --add \ --allow-principal User:order-app \ --operation Read --resource-pattern-type prefixed \ --group order- kafka-acls.sh $BS $AUTH --add \ --allow-principal User:admin \ --operation Create --operation ClusterAction \ --cluster kafka-acls.sh $BS $AUTH --add \ --allow-principal User:order-app \ --operation IdempotentWrite --cluster kafka-acls.sh $BS $AUTH --add \ --allow-principal User:order-app \ --operation Write \ --transactional-id order-tx- --resource-pattern-type prefixed kafka-acls.sh $BS $AUTH --list kafka-acls.sh $BS $AUTH --list --topic order-event kafka-acls.sh $BS $AUTH --remove --allow-principal User:order-app --topic order-event
11.4.3 通过 Broker 查询 ACL 1 2 3 kafka-acls.sh --bootstrap-server kafka1:9094 \ --command-config admin-sasl.properties --list
11.5 安全体系排障
现象
原因
解决
Authentication failed: Invalid username or password
SCRAM 用户不存在/密码错
使用可登录的管理员账号通过 --bootstrap-server 核对或重置;仅在 Broker 全部不可用的受控恢复场景使用已弃用的 --zookeeper 路径
javax.net.ssl.SSLHandshakeException: PKIX path building failed
客户端 truststore 缺少正确 CA
导入 CA 到 truststore
No subject alternative names matching IP address
证书 CN/SAN 与连接地址不匹配
证书 SAN 加 IP,或改用主机名连接
TopicAuthorizationException: Not authorized to access topics
ACL 配置缺失
按 11.4.2 补授权
GroupAuthorizationException
缺失 group 的 Read 权限
注意 group 权限是 Read ,不是 Write
Broker 之间连不上
security.inter.broker.protocol 与 JAAS 不匹配
检查双向认证的 Client 段配置
Super user not defined 警告
super.users 未配置
至少配置一个管理账号为 super user
加了 ACL 后自己也连不上
开启 ACL 时未把自己设为 super user
提前配置 super.users=User:admin
落地顺序(极其重要,顺序错了会把自己锁在门外) :
1 2 3 4 5 6 ① 先在测试环境验证 SASL+SSL 全链路 ② 生产先保留 PLAINTEXT 监听端口,新增 SASL_SSL 端口(双协议并存) ③ 配置好 super.users,确认管理员可用 ④ 业务方逐个切换到 SASL_SSL ⑤ 确认无 PLAINTEXT 连接后,关闭 PLAINTEXT 端口 ⑥ 开启 ACL(allow.everyone.if .no .acl.found =false )
第 12 章 集群扩容、缩容与迁移
扩容不是”加台机器就完事”。Kafka 扩容有三个动作:加 Broker → 数据重平衡 → 观察稳定 。 其中”数据重平衡”是最容易出事的一步(限速、Leader 切换、磁盘打满)。
12.1 扩容前评估 12.1.1 判断是否真的需要扩容
指标
阈值
说明
磁盘使用率
> 70%
需要扩磁盘或扩节点
单 Broker 分区数
> 2000
元数据/句柄压力大,需扩节点
网络出口带宽
> 70% 线速
需加网卡或扩节点
CPU 使用率(Kafka 进程)
持续 > 70%
需扩节点
生产 P99 延迟
持续 > 100ms
先调优,再考虑扩容
副本同步延迟
持续增长
优先排查而非扩容
12.1.2 扩容方案对比
方案
操作
适用
风险
加 Broker
新节点入集群 + 分区重分配
吞吐/存储不足
重分配耗时长,需限速
加磁盘(同节点)
log.dirs 加路径 + 重启
仅磁盘不足
需重启,且新盘不自动分担旧数据
加磁盘(扩分区)
云盘在线扩容 + xfs_growfs
云环境
无
垂直扩容
加 CPU/内存
短期应急
需重启
经验 :优先”加 Broker”而不是”加磁盘” 。因为加磁盘不改变分区分布(Kafka 不会自动把已存在的分区迁到新盘),单盘热点问题依旧;而加 Broker 后通过重分配可以彻底解决负载倾斜。
12.2 新增 Broker 完整流程(标准 SOP) 步骤 1:准备机器与环境 1 2 3 4 5 java -versioncat /etc/security/limits.conf | tail -5 sysctl -p | grep -E "swappiness|somaxconn" df -h /data
步骤 2:拷贝安装包(保持版本完全一致) 1 2 3 scp kafka1:/opt/kafka_2.13-3.6.2.tgz /tmp/cd /opt && tar -xzf /tmp/kafka_2.13-3.6.2.tgz && ln -s /opt/kafka_2.13-3.6.2 /opt/kafka
步骤 3:分配 broker.id(必须全局唯一且不复用) 1 2 3 4 5 6 /opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka ls /brokers/ids kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep '(id:'
重要 :避免复用已下线节点的 broker.id。技术上可以复用,但旧数据目录、监控标签和自动化台账容易产生歧义;建议 ID 单调递增。
步骤 4:配置 server.properties 1 2 3 4 5 6 7 8 9 broker.id =4 listeners =PLAINTEXT://kafka4:9092 advertised.listeners =PLAINTEXT://kafka4:9092 log.dirs =/data1/kafka,/data2/kafka zookeeper.connect =zk1:2181,zk2:2181,zk3:2181/kafka num.partitions =12 default.replication.factor =3 min.insync.replicas =2
1 2 3 scp kafka1:/opt/kafka/config/server.properties /tmp/server.ref diff /tmp/server.ref /opt/kafka/config/server.properties
步骤 5:创建数据目录并启动 1 2 3 4 5 6 7 8 mkdir -p /data1/kafka /data2/kafkachown -R kafka:kafka /data1 /data2 systemctl start kafkatail -f /opt/kafka/logs/server.log kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep -c '(id:'
此时注意 :新 Broker 不会自动接管任何分区 。它只是一个”空跑”的节点(除非新 Topic 创建)。必须做下面的重分配。
12.3 分区重分配(把负载搬到新节点) 12.3.1 生成重分配方案 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 cat > /tmp/topics-to-move.json <<'EOF' {"topics" : [ {"topic" : "order-event" }, {"topic" : "user-behavior" }, {"topic" : "log-collect" } ],"version" : 1} EOF kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --topics-to-move-json-file /tmp/topics-to-move.json \ --broker-list "1,2,3,4" --generate > /tmp/reassign-plan.txt sed -n '/Proposed partition reassignment configuration/,$p' /tmp/reassign-plan.txt \ | tail -n +2 | head -1 > /tmp/reassign.json
⚠️ 重要认知 :--generate 会重新洗牌所有分区 ,即使某些分区已经在新节点上,也可能被再次搬迁,造成不必要的巨量数据拷贝 。生产环境更稳妥的做法是手动编写 JSON ,只迁移需要的分区。
12.3.2 精准迁移(推荐做法:只迁部分分区) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 kafka-topics.sh --bootstrap-server kafka1:9092 --describe --topic order-eventcat > /tmp/reassign-target.json <<'EOF' { "version" : 1, "partitions" : [ {"topic" : "order-event" , "partition" : 0, "replicas" : [4, 2, 3]}, {"topic" : "order-event" , "partition" : 3, "replicas" : [4, 3, 2]} ] } EOF
生成脚本:自动按磁盘用量均衡迁移
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 #!/bin/bash python3 - <<'PY' import json plan = {"version" : 1, "partitions" : []}for p in [0, 2, 4, 6, 8, 10]: plan["partitions" ].append( {"topic" : "order-event" , "partition" : p, "replicas" : [4, 2, 3]} ) json.dump(plan, open("/tmp/reassign-target.json" , "w" ), indent=2)print ("generated:" , len(plan["partitions" ]), "partitions" ) PY
12.3.3 执行重分配(必须限速! ) 1 2 3 4 5 kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --reassignment-json-file /tmp/reassign-target.json \ --execute --throttle 52428800
如何设置合理的 throttle 值 :
1 2 3 4 throttle = min(带宽剩余量的 50%, 磁盘写入余量) 例如:万兆网卡(1250MB/s),日常使用 400MB/s → 剩余 850MB/s 取一半 → 约 400MB/s,但还要考虑副本同步流量 保守取 100MB/s = 104857600
12.3.4 观察进度 1 2 3 4 5 6 7 8 watch -n 30 'kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --reassignment-json-file /tmp/reassign-target.json --verify'
同时盯住这两个指标 (重分配期间的守护线):
1 2 3 4 5 kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions | wc -l iostat -x 1 | grep -A1 Device
12.3.5 清除限速(90% 的人会忘! ) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --reassignment-json-file /tmp/reassign-target.json --verify kafka-configs.sh --bootstrap-server kafka1:9092 --describe --broker 1 | grep -i throttl kafka-configs.sh --bootstrap-server kafka1:9092 --alter --broker 1 \ --delete-config follower.replication.throttled.rate kafka-configs.sh --bootstrap-server kafka1:9092 --alter --broker 1 \ --delete-config leader.replication.throttled.rate kafka-configs.sh --bootstrap-server kafka1:9092 --alter --topic order-event \ --delete-config follower.replication.throttled.replicas kafka-configs.sh --bootstrap-server kafka1:9092 --alter --topic order-event \ --delete-config leader.replication.throttled.replicasfor b in 1 2 3 4; do echo "--- broker $b ---" kafka-configs.sh --bootstrap-server kafka1:9092 --describe --broker $b | grep -i throttldone
12.3.6 重分配完成后:Leader 均衡 重分配只改了副本位置 ,Leader 仍是原来的(除非原 Leader 的副本位置变了)。需要触发 Preferred Leader 选举让 Leader 回到首选位置:
1 2 3 4 5 6 7 8 9 kafka-leader-election.sh --bootstrap-server kafka1:9092 \ --election-type PREFERRED --all-topic-partitions kafka-topics.sh --bootstrap-server kafka1:9092 --describe | \ awk '/Partition:/ {for(i=1;i<=NF;i++) if($i=="Leader:") print $(i+1)}' | \ sort | uniq -c
自动均衡更省事 :Kafka 3.x 支持 auto.leader.rebalance.enable=true(默认就是 true),配合 leader.imbalance.check.interval.seconds=300,会自动把不均衡的 Leader 拉回首选位置。但生产上很多人会关掉它(怕”自动操作”不可控),此时需要定期手动执行上面的命令 。
12.4 缩容 / 下线节点 12.4.1 标准下线流程 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 kafka-topics.sh --bootstrap-server kafka1:9092 --describe | grep -E "Leader: 4 " | wc -l kafka-topics.sh --bootstrap-server kafka1:9092 --describe | \ awk -v OFS="\n" '/Partition:/ {if ($0 ~ /Leader: 4 / || $0 ~ /Replicas: [^ ]*4/) print $0}' | head -50 python3 - <<'PY' import json, subprocess, re out = subprocess.check_output( ["/opt/kafka/bin/kafka-topics.sh" , "--bootstrap-server" , "kafka1:9092" , "--describe" ], text=True) plan = {"version" : 1, "partitions" : []}for line in out.splitlines(): m = re.match(r"\s*Topic: (\S+)\s+Partition: (\d+)\s+Leader: (\S+)\s+Replicas: (\S+)\s+Isr: (\S+)" , line) if not m: continue topic, part, leader, replicas = m.group(1), int(m.group(2)), m.group(3), m.group(4) rl = replicas.split("," ) if "4" not in rl: continue new = [b for b in ["1" ,"2" ,"3" ] if b not in rl][:1] + [b for b in rl if b != "4" ] plan["partitions" ].append({"topic" : topic, "partition" : part, "replicas" : [int(x) for x in new]}) json.dump(plan, open("/tmp/remove-broker4.json" ,"w" ), indent=2)print ("partitions to move:" , len(plan["partitions" ])) PY kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --reassignment-json-file /tmp/remove-broker4.json --execute --throttle 52428800 kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --reassignment-json-file /tmp/remove-broker4.json --verify kafka-topics.sh --bootstrap-server kafka1:9092 --describe | grep -c "Replicas: [^ ]*4" systemctl stop kafka /opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka ls /brokers/ids
12.4.2 下线注意事项
绝不能直接关机 !否则该节点上的所有 Leader 会被强制切换(虽然有多副本保护,但会造成短暂的元数据抖动和客户端重连)。
下线前确认 UnderReplicatedPartitions == 0。
下线过程中要监控其他节点的磁盘(接收数据会增长)。
保留下线节点的数据目录 至少 7 天 ,以防需要回滚。
下线后清理 ZK 中残留的配置:/config/brokers/4。
12.5 磁盘扩容与数据目录迁移 12.5.1 单机加磁盘(JBOD 模式) 1 2 3 4 5 6 7 8 9 10 11 mkfs.xfs /dev/sdcmkdir -p /data3echo "/dev/sdc /data3 xfs defaults,noatime 0 0" >> /etc/fstab mount -a log.dirs=/data1/kafka,/data2/kafka,/data3/kafka systemctl restart kafka
⚠️ 重要 :加新盘后,已有分区的数据不会自动迁到新盘 。要真正利用新盘空间,需要把部分分区迁到新盘:
1 使用 `kafka-reassign -partitions.sh` 的 `log_dirs` 字段在同一 Broker 内迁移副本目录,不需要先迁到其他 Broker。
同一 Broker 内迁移副本到新目录 :
1 2 3 4 5 6 7 8 9 10 11 { "version" : 1 , "partitions" : [ { "topic" : "order-event" , "partition" : 0 , "replicas" : [ 1 , 2 , 3 ] , "log_dirs" : [ "/data3/kafka" , "any" , "any" ] } ] }
1 2 3 4 5 6 7 kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --reassignment-json-file /tmp/move-log-dir.json \ --execute --replica-alter-log-dirs-throttle 52428800 kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \ --reassignment-json-file /tmp/move-log-dir.json --verify
12.5.2 单盘替换(磁盘故障) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 kafka-log-dirs.sh --bootstrap-server kafka1:9092 --broker-list 1 --describe | python3 -m json.tool systemctl stop kafka systemctl start kafka kafka-log-dirs.sh --bootstrap-server kafka1:9092 --broker-list 1 --describe | grep error
12.6 跨集群数据迁移(MirrorMaker 2) 12.6.1 为什么需要跨集群
机房搬迁 / 云迁移(自建 → 云)
异地容灾
数据汇聚(多集群 → 总部集群)
环境隔离(生产 → 测试)
12.6.2 MirrorMaker 2 核心概念
概念
说明
Source Connector
从源集群消费
Checkpoint Connector
同步消费者组 offset(迁移后消费者可续读)
Heartbeat Connector
心跳,用于监控链路延迟
Topic 重命名
默认 <source-alias>.<topic>(例如 source.orders),可配置 replication policy
自动创建 Topic
MM2 会自动在目标集群创建相同的 Topic(含配置)
12.6.3 MM2 部署(生产版) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 clusters = source, target source.bootstrap.servers = src-kafka1:9092,src-kafka2:9092,src-kafka3:9092 target.bootstrap.servers = dst-kafka1:9092,dst-kafka2:9092,dst-kafka3:9092 source->target.enabled = true source->target.topics = .* # 或精确列表 source->target.topics.exclude = ^__.*|^_schemas.* source->target.groups = .* source->target.sync.topic.configs.enabled = true source->target.sync.topic.acls.enabled = true source->target.producer.acks = all source->target.producer.enable.idempotence = true source->target.producer.compression.type = lz4 source->target.consumer.max.poll.records = 1000 source->target.consumer.auto.offset.reset = earliest source->target.emit.heartbeats.enabled = true source->target.emit.heartbeats.interval.seconds = 5 source->target.emit.checkpoints.enabled = true source->target.sync.group.offsets.enabled = true tasks.max = 8 # 建议 = 源集群分区数/10 replication.factor = 3 refresh.topics.interval.seconds = 60 refresh.groups.interval.seconds = 60 offset.flush.interval.ms = 10000
1 2 3 cd /opt/kafka bin/connect-mirror-maker.sh /opt/kafka/config/mm2.properties
12.6.4 MM2 运维要点 1 2 3 4 5 6 7 8 9 10 11 12 13 curl -s http://mm2-host:8083/connectors | python3 -m json.tool curl -s http://mm2-host:8083/connectors/source->target.MirrorSourceConnector/status | python3 -m json.tool
12.6.5 迁移切换(最关键的收尾动作) 1 2 3 4 5 6 7 8 T-7d :MM2 上线,双写前先单向同步(源 → 目标) T-3d :验证数据一致性(比对各 Topic 的 offset 差 / 抽样比对内容) T-0 :① 停止源集群生产者 ② 等待 MM2 积压归零(检查 replication-latency-ms) ③ 停止 MM2 ④ 业务切换到目标集群(改 bootstrap. servers) ⑤ 观察 24h ,源集群保留只读 T+7d :确认无误后,源集群下线
不可逆风险提示 :切换后如果发现问题要回退,源集群必须还有数据 !所以切流前不要 清理源集群数据,且回退窗口内不能有 retention 清理。
12.7 同机房整体搬迁(换机器) 目标 :把 3 台老机器上的集群,平滑搬到 3 台新机器上,业务无感知。
1 2 3 4 5 ① 新机器上部署 3 个 Broker(broker.id = 4 ,5 ,6 ),加入同一集群 ② 逐个把老节点的分区迁到新节点(按 12 .4 的方法,一次一个 broker) broker1 → 4 → 验证 → broker2 → 5 → 验证 → broker3 → 6 ③ 全部迁完后,下线老节点(broker 1 ,2 ,3 ) ④ (可选)把新节点的 broker.id 改回 1 ,2 ,3 ,但这会有风险,一般不建议
要点 :
每次迁移后必须等 UnderReplicatedPartitions == 0 再做下一个;
迁移期间全程限速;
老节点数据保留至少 7 天后才可清理。
第 13 章 版本升级与数据迁移
Kafka 的滚动升级是官方支持且相对安全 的,但升级前必须理解两个”版本闸门”参数:inter.broker.protocol.version 与 log.message.format.version。这是 90% 升级事故的根源。
13.1 升级路线与兼容性 13.1.1 支持的升级路径
当前版本
目标版本
是否支持
说明
2.8.x
3.6.2
✅
3.6 官方支持直接滚动升级,无需先到 3.0;仍须阅读跨版本变更
3.0
3.9
需核对目标版本指南
以 3.9 官方升级页列出的来源版本和前置条件为准
3.x
更高 3.x
需核对目标版本指南
不用经验表替代目标版本的升级说明
3.x → 4.x
❌
4.0 移除 ZK,需先迁 KRaft
官方兼容性矩阵(客户端/服务端) :
1 2 3 Kafka 客户端与 Broker 通过 API 版本协商实现跨版本兼容。 `log .message.format .version ` 控制磁盘消息格式,不决定全部客户端协议兼容性。 升级前应按目标版本兼容性说明核对生产者、消费者、Kafka Streams 和 Connect 。
13.1.2 升级前的检查清单 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | head -5 > /tmp/before-upgrade.txt kafka-topics.sh --bootstrap-server kafka1:9092 --describe > /tmp/topics-before.txt kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --all-groups --describe > /tmp/cg-before.txt kafka-configs.sh --bootstrap-server kafka1:9092 --describe --entity-type topics --all \ > /tmp/topic-configs-before.txt kafka-configs.sh --bootstrap-server kafka1:9092 --describe --entity-type brokers --all \ > /tmp/broker-configs-before.txt /opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka ls /brokers/ids kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions | wc -l kafka-topics.sh --bootstrap-server kafka1:9092 --describe --unavailable-partitions | wc -lcp -r /opt/kafka_2.13-3.6.2 /opt/kafka_2.13-3.6.2.bakecho "ruok" | nc zk1 2181 echo "stat" | nc zk1 2181 | head -5
13.2 滚动升级标准 SOP 13.2.1 升级流程图 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 ① 升级前准备(备份、留档、健康检查) ↓ ② 升级第一个 Broker(broker 1 ) · 停 → 替换二进制 → 启动(保持 inter.broker .protocol .version 不变) ↓ ③ 验证 broker 1 正常,观察 30 分钟 ↓ ④ 升级 broker 2 、broker 3 (同上) ↓ ⑤ 全部 Broker 完成二进制升级 ↓ ⑥ 二次滚动:修改 inter.broker .protocol .version 到新版本 (一次一台,滚动重启) ↓ ⑦ 验证集群全绿 ↓ ⑧ (可选)修改 log.message .format .version(需再次滚动重启) ↓ ⑨ 客户端逐个升级
13.2.2 详细操作(以 3.6.2 → 3.7.0 为例) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 cd /opt wget https://archive.apache.org/dist/kafka/3.7.0/kafka_2.13-3.7.0.tgz tar -xzf kafka_2.13-3.7.0.tgz kafka-topics.sh --bootstrap-server kafka1:9092 --describe | grep "Leader: 1 " | wc -l systemctl stop kafka ps -ef | grep -c "[k]afka.Kafka" rm -f /opt/kafkaln -s /opt/kafka_2.13-3.7.0 /opt/kafkacp /opt/kafka_2.13-3.6.2.bak/config/server.properties /opt/kafka/config/server.properties diff /opt/kafka/config/server.properties /opt/kafka_2.13-3.7.0/config/server.properties.newcat >> /opt/kafka/config/server.properties <<'EOF' inter.broker.protocol.version=3.6-IV2 EOF systemctl start kafkatail -f /opt/kafka/logs/server.log kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | head -3 kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions | wc -l
13.2.3 两个”版本闸门”参数详解(核心知识 )
参数
作用
默认值
何时改
inter.broker.protocol.version
ZooKeeper 模式下 Broker 间功能/协议级别
3.6.2 默认为 3.6-IV2
所有 Broker 二进制升级完成后再提升
log.message.format.version
写入磁盘的消息格式版本
3.6.2 默认为 3.0-IV1
现代 3.x 集群通常保持默认,不要写无效的 3.6
为什么必须先保持不变?
1 2 3 4 5 6 场景:3 台 Broker,把 broker1 升级到 3 .7 ,但未预先锁定 IBP → 新 Broker 可能启用目标版本功能级别 → 增加回退难度,并可能提前写入旧版本不能处理的元数据状态 正确做法:升级二进制前在所有 Broker 锁定 inter.broker.protocol.version=3.6-IV2 等 3 台全部升到 3 .7 后,再改成 3 .7 -IV4 并滚动重启
修改协议版本的操作 :
1 2 3 4 5 6 7 8 9 10 11 12 sed -i 's/^inter.broker.protocol.version=3.6-IV2$/inter.broker.protocol.version=3.7-IV4/' \ /opt/kafka/config/server.properties systemctl restart kafkasleep 300 kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions | wc -l kafka-broker-api-versions.sh --bootstrap-server kafka1:9092
注意 :inter.broker.protocol.version 需要滚动重启 才能生效,不能动态修改。
提示 :Kafka 3.6/3.7 的默认消息格式仍是 3.0-IV1。现代集群通常不需要显式设置 log.message.format.version;只有维护非常老的消费者兼容性时才按官方升级说明单独控制。
13.3 升级中的监控与验证 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 while true ; do echo "===== $(date '+%F %T') =====" echo -n "Broker 数: " kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 2>/dev/null | grep -c '(id:' echo -n "副本不足分区: " kafka-topics.sh --bootstrap-server kafka1:9092 --describe \ --under-replicated-partitions 2>/dev/null | grep -c "Partition:" echo -n "不可用分区: " kafka-topics.sh --bootstrap-server kafka1:9092 --describe \ --unavailable-partitions 2>/dev/null | grep -c "Partition:" echo -n "Controller: " /opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controller 2>/dev/null | \ grep -o '"brokerid":[0-9]*' sleep 30done
升级期间的红线 :
指标
红线
动作
UnderReplicatedPartitions
> 0 持续 5 分钟
暂停升级 ,等恢复
OfflinePartitions
> 0
立即回滚
生产 P99
> 500ms
暂停升级
Broker 数
少于预期
检查刚升级的节点是否启动成功
13.4 升级回退方案 1 2 3 4 5 6 7 8 9 systemctl stop kafkarm -f /opt/kafka && ln -s /opt/kafka_2.13-3.6.2 /opt/kafkacp /opt/kafka_2.13-3.6.2.bak/config/server.properties /opt/kafka/config/server.properties systemctl start kafka
铁律 :提升 inter.broker.protocol.version 后不能假设仍可简单降级。具体回退边界以目标版本官方升级指南为准;在确认新版本稳定前不要提升 IBP。
13.5 ZooKeeper → KRaft 迁移预备(3.x 特有的规划) Kafka 4.0 已移除 ZooKeeper 模式 。Kafka 3.6+ 提供 ZK → KRaft 的在线迁移流程,存量集群应在仍运行 3.x 时完成迁移。
迁移必须严格按目标 Kafka 版本的官方“ZooKeeper to KRaft Migration”流程执行。它不是 kafka-storage.sh format 加一次状态查询即可完成的操作;对存量 Broker 数据目录执行错误的 format 命令可能破坏元数据。
高层阶段为:准备独立 KRaft Controller 仲裁集群并启用迁移模式、等待元数据从 ZK 复制完成、滚动修改 Broker 连接 Controller、确认进入 KRaft 主导阶段、最后在满足回退窗口后退出迁移模式。每个阶段的配置项和回退边界随 Kafka 小版本变化,不能使用省略参数的通用命令。
运维建议 :
3.x + ZK 仅作为存量过渡架构维护 ,应明确迁移期限;
新集群直接使用当前受支持 Kafka 版本的 KRaft 模式;
若计划迁 KRaft,先在测试环境完整演练一次,记录每一步耗时与回退方式;
ZK 迁移的风险点在于元数据一致性 ,迁移后必须用 kafka-metadata-quorum.sh describe 与 kafka-topics.sh --describe 全面核对。
第 14 章 备份、容灾与恢复
残酷现实 :Kafka 本身没有”备份”功能 。它的”备份”就是多副本 。 所以,Kafka 的容灾体系 = 副本机制 + 跨集群复制 + 上游可重放 。
14.1 容灾体系全景
层级
手段
RPO
RTO
成本
L1 副本容灾
同集群 3 副本(跨机架)
0
秒级
低(3x 存储)
L2 集群容灾
MM2 跨集群同步
秒~分钟
分钟级
中
L3 同城双活
两集群 + MM2 双向
秒级
分钟级
高
L4 异地灾备
异地集群 + MM2 单向
分钟级
十分钟级
高
L5 数据重放
上游(DB binlog / 埋点)重灌
取决于上游
小时级
低(需上游支持)
务实建议 :绝大多数团队只需做到 L1 + L5 :本地 3 副本 + 保证上游可重放。MM2 双集群的成本(翻倍资源 + 运维复杂度)要用业务价值来论证。
14.2 元数据清单与 ZooKeeper 一致性备份 命令行导出适合审计和重建,但不是可直接导入的完整备份。ZooKeeper 模式集群还必须对 dataDir 与 dataLogDir 做一致性快照;消费位点位于 __consumer_offsets,仅导出 --describe 结果不能无损恢复。
14.2.1 备份内容清单
内容
存储位置
备份方式
Topic 列表与分区/副本定义
ZK /brokers/topics
kafka-topics.sh --describe 导出,供重建参考
Topic 配置
ZK /config/topics
按 --entity-type topics --all 导出
Broker 动态配置
ZK /config/brokers
按 --entity-type brokers --all 导出
ACL
ZooKeeper(路径取决于 chroot)
kafka-acls.sh --list 导出,供重建参考
消费组位点
__consumer_offsets Topic
跨集群用 MM2 Checkpoint;命令导出仅供审计
SCRAM 用户
ZK /config/users
按已批准用户名逐个导出,秘密值不可从 Broker 反查
Broker server.properties
各节点本地
scp 收集
ZooKeeper 数据
dataDir + dataLogDir
使用存储快照或停机冷备,保证两者时间点一致
14.2.2 元数据备份脚本(每日执行) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 #!/bin/bash set -euo pipefail BS="kafka1:9092,kafka2:9092,kafka3:9092" ZK="zk1:2181" KAFKA_BIN="/opt/kafka/bin" BACKUP_DIR="/data/backup/kafka-metadata/$(date +%Y-%m-%d) " mkdir -p "$BACKUP_DIR " echo "[$(date '+%F %T') ] 开始备份 Kafka 元数据到 $BACKUP_DIR " $KAFKA_BIN /kafka-topics.sh --bootstrap-server $BS --describe \ > "$BACKUP_DIR /topics-describe.txt" 2>&1$KAFKA_BIN /kafka-configs.sh --bootstrap-server $BS --describe \ --entity-type topics --all > "$BACKUP_DIR /topic-configs.txt" 2>&1$KAFKA_BIN /kafka-configs.sh --bootstrap-server $BS --describe \ --entity-type brokers --all \ > "$BACKUP_DIR /broker-configs.txt" 2>&1$KAFKA_BIN /kafka-acls.sh --bootstrap-server $BS --list \ > "$BACKUP_DIR /acls.txt" 2>&1 || true $KAFKA_BIN /kafka-consumer-groups.sh --bootstrap-server $BS --all-groups --describe \ > "$BACKUP_DIR /consumer-groups.txt" 2>&1$KAFKA_BIN /kafka-consumer-groups.sh --bootstrap-server $BS --all-groups \ --describe --state > "$BACKUP_DIR /cg-state.txt" 2>&1if command -v /opt/zookeeper/bin/zkCli.sh >/dev/null 2>&1; then /opt/zookeeper/bin/zkCli.sh -server $ZK <<EOF > "$BACKUP_DIR/zk-paths.txt" 2>&1 ls -R /kafka quit EOF fi for h in kafka1 kafka2 kafka3; do scp -q $h :/opt/kafka/config/server.properties "$BACKUP_DIR /server.properties.$h " || \ echo "WARN: 无法从 $h 复制配置" done tar czf "$BACKUP_DIR .tar.gz" -C "$(dirname $BACKUP_DIR) " "$(basename $BACKUP_DIR) " rm -rf "$BACKUP_DIR " find /data/backup/kafka-metadata -name "*.tar.gz" -mtime +30 -deleteecho "[$(date '+%F %T') ] 备份完成:$BACKUP_DIR .tar.gz"
上述脚本产生的是元数据清单 。另行对 ZooKeeper 的 dataDir 与 dataLogDir 做同一时间点的存储快照;若没有快照能力,应在维护窗口停止整个 ZooKeeper 集群后冷备,恢复流程必须先在测试环境演练。
1 2 0 3 * * * /opt/kafka/bin/backup-metadata.sh >> /var/log/kafka-backup.log 2>&1
14.2.3 从元数据备份恢复 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 awk '/^Topic: / && /PartitionCount:/ { for (i=1; i<=NF; i++) { if ($i=="Topic:") topic=$(i+1) if ($i=="PartitionCount:") partitions=$(i+1) if ($i=="ReplicationFactor:") rf=$(i+1) } print topic, partitions, rf }' /data/backup/kafka-metadata/2026-09-01/topics-describe.txt | sort -u | \while read topic partitions rf; do echo "重建 Topic: $topic (partitions=$partitions , rf=$rf )" /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka1:9092 \ --create --topic "$topic " \ --partitions "$partitions " \ --replication-factor "$rf " 2>/dev/null || echo " 已存在或创建失败,请核对" done
强烈建议 :在测试环境完整演练一次”从备份重建集群” 。没演练过的备份 = 没有备份。
14.3 数据备份与恢复 14.3.1 为什么”不备份数据文件”
原因
说明
数据量大
动辄 TB 级,全量拷贝成本极高
在线拷贝危险
拷贝过程中段文件在滚动,副本不一致
已有副本
3 副本本身就是 3 份数据
恢复慢
从备份恢复 TB 数据远比从副本同步慢
14.3.2 正确的数据保障策略 1 2 3 4 5 6 ① 副本数 = 3 (跨机架分布) ② min.insync.replicas = 2 + acks = all ③ 上游可重放(最重要!) · MySQL → Debezium 采集 binlog → Kafka(binlog 保留 7 天以上) · 业务埋点 → 本地落盘 buffer → Kafka(可重投) ④ 关键 Topic 单独做 MM2 到备份集群(仅限金融级核心数据)
14.3.3 特殊情况:离线拷贝数据文件 只在整个 Broker 停机 的前提下才安全:
1 2 3 4 systemctl stop kafka tar czf /backup/kafka-data-$(date +%F).tgz /data1/kafka /data2/kafka systemctl start kafka
即使冷备,恢复时必须保证 broker.id、topic、partition、offset 完全对应 ,否则数据无法被识别。这也是为什么实际运维中几乎不做数据文件备份。
14.4 容灾架构设计 14.4.1 方案一:同城双集群(推荐大多数场景) 1 2 3 4 5 6 7 8 9 10 11 12 ┌─────────────────┐ ┌─────────────────┐ │ 集群 A (主) │ │ 集群 B (备) │ │ 3 Broker │ │ 3 Broker │ └────────┬────────┘ └────────┬────────┘ │ │ │ MM2 单向同步 │ └───────────────────────────►┘ (主→备,延迟 < 1s ) 业务写:集群 A 业务读:集群 A 故障切换:客户端改 bootstrap.servers → 集群 B
要点 :
MM2 单向即可,双向会引入循环复制(需 Topic 重命名策略);
备集群的机器配置不应低于 主集群(否则故障时扛不住);
定期演练切换 (至少每季度一次);
消费位点同步依赖 Checkpoint Connector,务必开启。
14.4.2 方案二:双活(成本高,慎选) 1 2 集群 A ◄──── MM2 ────► 集群 B (业务 X 写 A ) (业务 Y 写 B )
复杂点 :两个方向的同步会产生同名 Topic 冲突 ,需通过 replication.policy.class 做 Topic 重命名,或用 DefaultReplicationPolicy 加集群前缀。运维复杂度高,除非业务确实需要,否则不建议 。
14.4.3 方案三:异地灾备(跨城) 1 2 3 同城集群(主) ──MM2 ──► 异地集群(灾备) │ │ 业务实时读写 仅同步,不承载业务
要点 :
跨城带宽是瓶颈,必须限速 + 压缩;
只同步核心 Topic ,不是全量;
RPO 通常是分钟级(受带宽限制);
做好”异地集群容量不足”的预案(故障时业务量全部压过来)。
14.5 容灾演练(必须定期做) 14.5.1 演练清单
演练项
频率
验证目标
单 Broker 宕机
月
多副本自动切换,业务无感
单机架断电(模拟)
季
跨机架副本起作用
Controller 切换
季
元数据操作短暂失败但自动恢复
ZK 单节点宕机
季
ZK 多数派仍可用
全集群故障 + 切换到备集群
半年
RTO 达标
从元数据备份重建集群
半年
备份真的可用
14.5.2 单 Broker 宕机演练脚本 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 #!/bin/bash BS="kafka1:9092" TARGET_BROKER=3echo "===== 演练前状态 =====" kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions | wc -lecho "===== 模拟 broker $TARGET_BROKER 宕机(kill -9)=====" ssh kafka$TARGET_BROKER "kill -9 \$(pgrep -f kafka.Kafka)" START=$(date +%s)while true ; do UR=$(kafka-topics.sh --bootstrap-server $BS --describe \ --under-replicated-partitions 2>/dev/null | grep -c "Partition:" ) NOW=$(date +%s) echo "[$((NOW-START) )s] UnderReplicatedPartitions = $UR " [ "$UR " -eq 0 ] && [ $((NOW-START)) -gt 5 ] && break sleep 5 [ $((NOW-START)) -gt 300 ] && echo "超时!" && break done echo "===== 恢复 broker $TARGET_BROKER =====" ssh kafka$TARGET_BROKER "systemctl start kafka"
验收标准 :
杀死 Broker 的瞬间,其上的 Leader 在 10 秒内 完成切换;
生产/消费无失败(客户端有重试机制);
Broker 恢复后,ISR 在合理时间内 补齐(取决于数据量和限速设置)。
14.6 恢复优先级(出事时的取舍) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 P 0 (1 小时内必须恢复) ├─ 恢复 Broker 进程(让服务可用) └─ 恢复 ZK 集群(Kafka 的元数据依赖)P 1 (当天恢复) ├─ 恢复副本数到 3 (消除单点) └─ 恢复 Controller 稳定P 2 (本周恢复) ├─ 补齐落后的副本 └─ 处理历史积压P 3 (长期) ├─ 补齐丢失的数据(从上游重放) └─ 复盘与预防
核心原则 :先恢复可用性,再恢复完整性 。不要为了”不丢一条消息”而让整个业务停摆。
14.7 数据一致性核对工具 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 total_latest () { kafka-run-class.sh kafka.tools.GetOffsetShell \ --bootstrap-server kafka1:9092 --topic "$1 " --time -1 | \ awk -F: '{s+=$3} END {print s}' }echo "order-event latest offset watermark: $(total_latest order-event) " kafka-console-consumer.sh --bootstrap-server src:9092 --topic t1 \ --from-beginning --property print.key=true --property print.timestamp=true kafka-console-consumer.sh --bootstrap-server dst:9092 --topic source.t1 \ --from-beginning --property print.key=true --property print.timestamp=true kafka-dump-log.sh --files /data1/kafka/order-event-0/00000000000000000000.log \ --print-data-log --deep-iteration | tail -5
对压缩 Topic,offset 差值也不是现存消息数;跨集群验收应以业务记录数、唯一 ID/校验和、MM2 延迟和 checkpoint 状态共同判断。
第 15 章 ZooKeeper 运维专项
为什么 ZK 模式要单独一章? 因为在这套架构里,ZK 是 Kafka 的”心脏起搏器” :
Broker 注册与发现(/brokers/ids)
Controller 选举(/controller)
Topic / 分区元数据(/brokers/topics)
配置、ACL、配额
消费者位移(旧版,3.x 已迁到 __consumer_offsets)
ZK 挂了,Kafka 不会立刻死(已运行的读写还能继续),但一旦发生 Broker 重启或 Leader 切换,整个集群就会失去自愈能力。
15.1 ZooKeeper 在 Kafka 中的角色
ZK 节点路径
内容
影响的 Kafka 行为
/brokers/ids/<id>
每个在线 Broker 的临时节点(含 host/port)
Broker 存活探测、客户端路由
/brokers/topics/<topic>
Topic 的分区副本分配方案
Topic 元数据,创建/删除
/brokers/topics/<topic>/partitions/<n>/state
分区状态(Leader / ISR / epoch)
Leader 选举结果
/controller
当前 Controller 的 broker.id(临时节点)
集群管理中枢
/controller_epoch
Controller 任期编号(每次切换 +1)
防止脑裂的 fencing 机制
/admin/delete_topics
待删除 Topic 列表
Topic 删除异步执行
/admin/reassign_partitions
进行中的分区重分配
重分配进度状态
/admin/preferred_replica_election
待执行的 Preferred Leader 选举
Leader 均衡
/config/topics /config/brokers /config/users
动态配置 + SCRAM 凭据
动态改配置
/kafka-acl
ACL 规则(ACL 存储在 ZK)
鉴权
/cluster/id
集群唯一 ID
防止数据目录串用
/isr_change_notification
ISR 变更通知
触发客户端元数据刷新
查看这些节点 :
1 2 3 4 5 6 7 8 9 10 11 12 13 zkCli.sh -server zk1:2181/kafkals /brokers/ids get /brokers/ids/1 ls /brokers/topics get /brokers/topics/order-event get /controller get /controller_epoch ls /admin/delete_topics get /admin/reassign_partitions ls /config/topics get /cluster/id
运维禁忌 :永远不要用 zkCli.sh 手动 delete 或 set Kafka 的节点 (除极少数官方文档明确的场景)。这会直接破坏集群元数据一致性,导致不可恢复的故障。
15.2 ZooKeeper 四字命令实战 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 echo "ruok" | nc zk1 2181 echo "stat" | nc zk1 2181 echo "srvr" | nc zk1 2181 echo "conf" | nc zk1 2181 echo "mntr" | nc zk1 2181echo "cons" | nc zk1 2181 echo "dump" | nc zk1 2181 echo "wchs" | nc zk1 2181 echo "wchc" | nc zk1 2181 echo "envi" | nc zk1 2181echo "srst" | nc zk1 2181
四字命令被禁用时的处理 :ZK 3.5+ 默认只允许白名单命令。若 ruok 等返回 whitelist 相关错误,需在 zoo.cfg 加:
1 4lw.commands.whitelist =ruok,stat,srvr,mntr,conf,cons,dump,wchs,envi
15.3 ZooKeeper 监控指标与告警基线
指标
正常范围
告警阈值
说明
zk_avg_latency
< 5ms
> 20ms
ZK 响应变慢,会拖慢 Kafka 元数据操作
zk_max_latency
< 50ms
> 200ms
出现毛刺需关注 GC
zk_outstanding_requests
0
> 10 持续 1 分钟
ZK 请求积压,通常是磁盘慢
zk_znode_count
视规模
突增 > 20%
Kafka 元数据异常增长
zk_watch_count
视规模
突增
客户端 watch 泄漏
zk_open_file_descriptor_count
< 上限 70%
> 80%
需调大 ulimit
zk_num_alive_connections
稳定
突降
客户端断连
zk_synced_followers
= 集群数-1
少于预期
follower 掉线
ZK 侧的 Prometheus 监控(见 8.7) ,并配置如下告警:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 groups: - name: zookeeper rules: - alert: ZKLatencyHigh expr: zk_avg_latency > 20 for: 5m labels: {severity: warning } annotations: summary: "ZK 平均延迟过高 ({{ $value }} ms) —— 会影响 Kafka 元数据操作" - alert: ZKOutstandingRequests expr: zk_outstanding_requests > 10 for: 2m labels: {severity: critical } annotations: summary: "ZK 请求积压 {{ $value }} 个 —— 检查磁盘与 GC" - alert: ZKNoLeader expr: count(zk_server_state{state="leader"}) != 1 for: 1m labels: {severity: critical } annotations: summary: "ZK 集群没有 leader 或多个 leader —— 集群异常" - alert: ZKSyncedFollowersLow expr: zk_synced_followers < 2 for: 5m labels: {severity: warning } annotations: summary: "ZK follower 同步数不足"
15.4 ZK 会话机制与 Kafka 的关键耦合 1 2 3 4 5 6 7 8 Kafka Broker ── 建立 ZK 会话(zookeeper.session.timeout .ms,默认 18000 ms) │ ├─ ZooKeeper 客户端按协商后的会话超时安排 ping,不应按 tickTime/2 推算 │ └─ 若会话超时 → ZK 删除该 Broker 的临时节点 /brokers/ids/<id > → Controller 感知到 Broker 掉线 → 触发该 Broker 上所有分区 Leader 的重新选举 → 大规模元数据变更、客户端元数据刷新、可能的短暂不可用
关键结论 :
现象
原因
处置
Broker 进程活着,但被 ZK 判定掉线
GC 停顿 / 磁盘 hang > zookeeper.session.timeout.ms
加大该值 + 优化 GC
整个集群同时掉线
ZK 集群故障
恢复 ZK(见 15.6)
Controller 频繁切换
会话抖动
检查网络与 ZK 负载
推荐的会话超时配置 :
1 2 3 4 5 6 7 8 9 10 11 12 zookeeper.session.timeout.ms =18000 # 默认 18000;GC 停顿大的环境可加到 30000 zookeeper.connection.timeout.ms =18000 # 默认 null,未设置时使用 session timeout;按网络环境显式设置 tickTime =2000 initLimit =10 # 2000 × 10 = 20s,follower 初始同步上限 syncLimit =5 # 2000 × 5 = 10s,follower 同步上限
常见坑 :把 Kafka 的 zookeeper.session.timeout.ms 设成 60000,但 ZK 的 maxSessionTimeout 只有 40000。ZooKeeper 会协商到允许范围内的值,服务端日志会记录协商结果;排障时应核对实际会话超时,而不是只看客户端配置文件。
15.5 ZooKeeper 常见故障处理 15.5.1 ZK 单节点宕机 影响 :若为 follower,集群仍可服务(多数派仍在),Kafka 无感知。若为 leader,触发 leader 重选举(通常几秒内完成)。
1 2 3 4 5 6 7 8 9 for h in zk1 zk2 zk3; do echo "--- $h ---" ; echo "srvr" | nc -w 2 $h 2181 | grep -E "Mode|Zxid" ; done systemctl start zookeeperecho "srvr" | nc zk1 2181echo "stat" | nc zk1 2181 | grep -E "Mode|Zxid|Connections"
要点 :ZK 的 follower 重启后会自动从 leader 同步,不需要手动干预 。但如果宕机时间过长(超过快照+事务日志保留范围),需要全量同步 ,会消耗较多带宽/时间。
15.5.2 ZK 失去多数派(脑裂风险) 场景 :3 节点 ZK 挂了 2 个,只剩 1 个。
1 2 3 echo "srvr" | nc zk1 2181 | grep Mode
处置(按优先级) :
尽快恢复至少 1 个节点 ,让多数派恢复(3 节点需要 2 个);
若节点数据丢失,从备份恢复快照 + 事务日志;
绝不要 通过修改 zoo.cfg 的 server 列表来”强行选主”——这会破坏 ZK 的一致性,可能导致数据分裂。
ZK 一致性公式 :可用节点数 ≥ (总节点数 / 2) + 1。3 节点可挂 1 个,5 节点可挂 2 个。这就是 ZK 必须部署奇数的原因。
15.5.3 ZK 磁盘满 1 2 df -h /data/zookeeperdu -sh /data/zookeeper/version-2/* | sort -rh | head
原因 :快照与事务日志未清理(autopurge 未开启或失效)。
1 2 3 autopurge.snapRetainCount =10 # 保留 10 个快照 autopurge.purgeInterval =24 # 每 24 小时清理一次
1 2 /opt/zookeeper/bin/zkCleanup.sh -n 10
禁止 按文件时间分别删除 snapshot.* 与 log.*。两类文件共同构成恢复链,应使用 autopurge 或 zkCleanup.sh/PurgeTxnLog 处理。
15.5.4 ZK 延迟高、请求积压 原因排查顺序 :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 jstat -gcutil $(pgrep -f QuorumPeerMain) 1000 5 iostat -x 1 | grep -A1 Deviceecho "mntr" | nc zk1 2181 | grep znode_countecho "mntr" | nc zk1 2181 | grep watch_count grep -cE 'TopicChange|ConfigChange|ReassignPartitions' /opt/kafka/logs/controller.log
优化手段 :
1 2 3 4 5 6 7 8 9 tickTime =2000 dataLogDir =/data/zk-txlog maxClientCnxns =200 admin.enableServer =false
1 2 3 4 5 export SERVER_JVMFLAGS="-Xms4g -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=200 \ -XX:+ParallelRefProcEnabled -XX:+HeapDumpOnOutOfMemoryError \ -XX:HeapDumpPath=/var/log/zookeeper/heapdump.hprof \ -Xlog:gc*:file=/var/log/zookeeper/gc.log:time:filecount=5,filesize=50M"
15.6 ZooKeeper 数据清理与快照管理 1 2 3 4 5 ls -lht /data/zookeeper/version-2/ /data/zk-txlog/version-2/ | head -40 /opt/zookeeper/bin/zkCleanup.sh -n 5
1 2 0 4 * * * /opt/zookeeper/bin/zkCleanup.sh -n 5 >> /var/log/zk-cleanup.log 2>&1
重要 :优先使用 autopurge。人工清理前应完成 dataDir 与 dataLogDir 的一致性备份,并在测试环境验证恢复。
15.7 ZooKeeper 集群运维 15.7.1 ZK 扩容(3 → 5 节点) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 server.1=zk1:2888:3888 server.2=zk2:2888:3888 server.3=zk3:2888:3888 server.4=zk4:2888:3888 server.5=zk5:2888:3888for h in zk1 zk2 zk3; do echo "srvr" | nc $h 2181 | grep Mode ssh $h "systemctl restart zookeeper && sleep 10" for z in zk1 zk2 zk3; do echo "srvr" | nc -w 2 $z 2181 | grep -E "^Mode" ; done done ssh zk4 "systemctl start zookeeper" ssh zk5 "systemctl start zookeeper" for h in zk1 zk2 zk3 zk4 zk5; do echo -n "$h : " ; echo "srvr" | nc -w 2 $h 2181 | grep Mode; done
扩容方向不可逆 :ZK 支持扩到 5、7 节点,但不建议缩容(缩容需要重新格式化数据目录,风险极高)。
15.7.2 ZK 换机器(IP 不变或变) IP 不变 :直接下线老节点、上新节点、改 myid 即可。IP 变化 :需在所有节点的 zoo.cfg 中改 server 列表并滚动重启,同时确认 Kafka 的 zookeeper.connect 同步更新。
1 2 zookeeper.connect =zk1:2181,zk2:2181,zk3:2181/kafka
15.7.3 ZK 版本升级 ZK 跨大版本升级(如 3.4 → 3.8)需谨慎,且 Kafka 对其 ZK 版本有兼容要求:
Kafka 版本
兼容的 ZK 版本
3.0 ~ 3.3
3.5.x ~ 3.6.x
3.4 ~ 3.6
3.6.x ~ 3.8.x
3.7 ~ 3.9
3.7.x ~ 3.9.x
生产建议:ZK 版本不要比 Kafka 官方推荐版本更新 ,避免未验证的兼容性问题。
15.8 ZooKeeper 运维红线清单
❌ 绝对不要做的事
后果
用 zkCli 手动 create/delete/set Kafka 的 znode
元数据损坏,集群不可恢复
生产环境执行 kill(四字命令)
ZK 立即退出
部署偶数个 ZK 节点
无法形成稳定多数派
把 ZK 与 Kafka 部署在同一台机器
资源争抢,故障域重叠
数据目录与事务日志目录放在同一块盘
ZK 延迟高,影响整个 Kafka 集群
删除 ZK 数据不做备份
ZK 无法恢复,Kafka 元数据丢失
使用 zookeeper.connect 不带 chroot(/kafka)
与其他应用共用 ZK 时数据混淆
忽略 zk_outstanding_requests > 0
小问题拖成集群级故障
第 16 章 常见生产事故案例复盘
本章收录 15 个真实世界的典型事故。每一个都是血泪换来的 。 建议团队把这一章作为新人培训材料,逐条对照自己的集群做检查。
案例 1:磁盘写满导致集群雪崩
项
内容
现象
3 台 Broker 中的 2 台磁盘 100%,随后所有生产者超时,业务全线告警
根因
某新上线的日志 Topic 未设置 retention.bytes,retention.ms 为 7 天,单日写入 2TB,磁盘 3 天写满
连锁反应
磁盘满 → Broker 无法写入 → 这些 Broker 上的 Leader 全部不可写 → ISR 收缩 → acks=all 的生产者全部阻塞 → 消息积压 → 内存 OOM
止血
①立即缩短该 Topic 的 retention.ms 到 1 小时;②临时把 log.retention.check.interval.ms 调小到 60s 触发清理;③对积压的生产者降级
根治
①所有 Topic 必须同时设置 retention.ms 和 retention.bytes;②建立磁盘容量预测模型(见 2.1);③磁盘 70% 告警、85% 严重告警
预防
Topic 创建流程中加入”必须指定保留策略”的强制校验
1 2 3 每日数据量 = 峰值 TPS × 平均消息大小 × 86400 × 压缩比⁻¹ 所需磁盘 = 每日数据量 × 保留天数 × 副本数 ÷ 0.7
案例 2:ISR 收缩 + acks=all 导致生产完全阻塞
项
内容
现象
业务方突然全部报”发送超时”,持续 20 分钟
时间线
14:02 一台 Broker 因磁盘 hang 掉线 → 14:03 其上分区 ISR 从 3 缩到 2 → 若有分区 ISR 缩到 1,min.insync.replicas=2 不满足 → 生产报 NotEnoughReplicasException
根因
①Broker 磁盘 I/O hang(老 SATA 盘,await 200ms+);②min.insync.replicas=2 与副本数 3 的组合下,一旦 ISR 缩到 1 就完全不可写
止血
①强制重启故障 Broker(虽然会丢一点未刷盘数据);②本次紧急将 min.insync.replicas 临时改为 1(有丢数据风险,需业务确认)
根治
①更换 SSD;②副本数保持 3,min.insync.replicas=2 不改(可靠性优先);③提高 replica.lag.time.max.ms 到 45s,减少误判导致的 ISR 收缩
预防
①UnderReplicatedPartitions 立即告警;②磁盘 await > 20ms 告警
权衡思考 :min.insync.replicas=2 是”可用性换可靠性”的典型。它的正确含义是:宁可不可写,也不接受可能丢数据的写入 。如果业务不能接受不可写,就增加副本数(如 4 副本 + min.insync=2)。
案例 3:残留复制限速导致副本永远追不上
项
内容
现象
一次扩容后,UnderReplicatedPartitions 长期为 30 左右,持续 3 天不恢复
排查
Broker 都在线、磁盘正常、网络正常;最后在 kafka-configs.sh --describe --broker 2 中发现 follower.replication.throttled.rate=1048576(1MB/s)
根因
3 天前做分区迁移时设置了 1MB/s 限速(当时带宽紧张),迁移完成后忘记清除 。此后所有副本同步都被限速在 1MB/s,1TB 的分区需要 11 天才能同步完
止血
清除所有 Broker 和 Topic 上的限速配置,副本在 15 分钟内全部追平
预防
①把”清除限速”写进迁移 SOP 的最后一步 ;②限速配置纳入每日巡检(见附录 C 脚本);③迁移脚本中加入 trap 保证退出时清限速
1 2 3 4 5 6 for b in $(kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | \ sed -n 's/.*(id: \([0-9]\+\).*/\1/p' ); do RES=$(kafka-configs.sh --bootstrap-server kafka1:9092 --describe --broker "$b " 2>/dev/null | grep -i throttl) [ -n "$RES " ] && echo "⚠️ Broker $b 存在残留限速: $RES " done
案例 4:ZK 会话超时导致 Broker 假死 + 大规模 Leader 切换
项
内容
现象
凌晨 3:00 突然出现大量 Leader 切换,持续 5 分钟,部分业务报”分区不可用”
根因
Broker 堆 8G,凌晨有日志清理 + 备份任务抢占 IO,导致一次 Full GC 停顿 22 秒 > zookeeper.session.timeout.ms=18000,ZK 判定 Broker 掉线,删除临时节点,Controller 触发其上 800 个分区的 Leader 重选举
止血
自动恢复(GC 结束后 Broker 重新注册,ISR 在 8 分钟内补齐)
根治
①GC 参数改为 G1GC + MaxGCPauseMillis=200,Full GC 降到 1s 内;②zookeeper.session.timeout.ms 提到 30000(需同步调大 ZK 的 maxSessionTimeout);③把 Kafka 数据盘与备份任务的盘分离
预防
①监控 GC 停顿,超过 3s 告警;②Broker 所在机器不跑其他重 IO 任务;③OOMScoreAdjust=-500
案例 5:消费者 Rebalance 风暴
项
内容
现象
某消费组消费停滞 2 小时,LAG 从 0 涨到 800 万
根因
消费者 max.poll.interval.ms 用默认 300s,但单批 max.poll.records=2000,单批处理耗时高峰时超过 300s(下游 DB 慢查询),消费者被踢出组 → Rebalance → 重新分配 → 又超时 → 循环 Rebalance
止血
临时把 max.poll.records 从 2000 降到 100,立即恢复
根治
①max.poll.records=200(与控制单批耗时);②消费逻辑改为异步处理 + 水位控制;③优化下游慢查询
预防
①监控日志中的 Rebalance 次数,超过阈值告警;②消费组上线前做压测,验证单批耗时
Rebalance 次数监控 :
1 2 3 4 kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group g1 | head -1 grep -c "Rebalancing" /opt/kafka/logs/server.log
案例 6:单分区热点导致单 Broker 打满
项
内容
现象
3 副本集群,broker 2 的 CPU 和磁盘 100%,其他两台只有 30%
根因
某 Topic 只有 3 个分区,其中 1 个分区承接了 80% 的流量(key 使用了少数几个”热门值”,如默认 key 或某大客户 ID),该分区 Leader 恰好在 broker 2
止血
迁移该分区的 Leader 到负载低的 Broker(治标)
根治
①增加分区数到 24(但需评估 key 分布);②生产端改进 key 设计(加盐值打散);③对超大客户单独建 Topic
预防
①监控单分区 的写入速率(不只是 Topic 总量);②大流量 Topic 分区数 ≥ 12
监控单分区速率 :
1 2 3 4 5 6 7 for i in 1 2; do kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server kafka1:9092 \ --topic order-event --time -1 > /tmp/off$i .txt sleep 60done paste /tmp/off1.txt /tmp/off2.txt | awk -F'[: \t]+' '{printf "%s-%s: %.2f msg/s\n",$1,$2,($6-$3)/60}'
案例 7:分区迁移未限速导致网络打满
项
内容
现象
扩容迁移开始后 10 分钟,全部业务生产延迟从 20ms 飙到 5s
根因
迁移命令未加 --throttle,全速拷贝 15TB 数据,把万兆网卡打满
止血
立即用 --execute --throttle <更低值> 下调限速;若必须停止,使用同一 JSON 执行 --cancel。--verify 只负责核对状态,不能取消进行中的迁移
根治
重新规划迁移,限速 100MB/s,分批进行(每次 3~5 个分区)
预防
①迁移命令模板强制带 --throttle;②迁移前评估带宽余量;③迁移期间监控网卡利用率,超过 70% 立即降低限速
案例 8:升级未锁定协议版本导致回退窗口失效
项
内容
现象
滚动升级 3.5 → 3.7 时未固定旧 IBP,升级中出现兼容性异常,且无法确认是否还能直接回退
根因
升级前未按目标版本升级指南锁定旧的 inter.broker.protocol.version,新 Broker 可能提前提升功能级别,增加混跑和回退风险
止血
暂停后续升级,保存日志和配置;按 3.7 官方升级指南核对当前 IBP 与回退边界后再决定继续升级或回退
根治
重新按正确流程升级:升级二进制时显式锁定 inter.broker.protocol.version,全部升级完成后再统一改版本
预防
①把”锁定协议版本”写进升级 SOP 的必做项;②升级前在测试环境完整演练;③升级脚本中加入自动校验(检测 inter.broker.protocol.version 是否与集群当前版本一致)
1 2 3 4 5 6 7 8 EXPECTED_OLD_IBP="3.5-IV2" CONFIG_IBP=$(awk -F= '/^inter.broker.protocol.version=/{print $2}' \ /opt/kafka/config/server.properties)if [ "$CONFIG_IBP " != "$EXPECTED_OLD_IBP " ]; then echo "协议版本未锁定为 $EXPECTED_OLD_IBP ,停止升级" exit 1fi
案例 9:OOM Killer 杀死 Broker
项
内容
现象
Broker 进程”莫名消失”,无异常日志尾部
根因
Broker 堆 8G + PageCache 占用大量内存 + 同机部署了其他内存消耗型服务(如 ZK、监控 agent),宿主机内存耗尽,OOM Killer 按 OOM Score 杀掉了内存占用最大的 Kafka
止血
重启 Kafka(数据不会丢,多副本保障)
根治
①Kafka 专用机器,不再混部;②OOMScoreAdjust=-500;③堆内存不超过物理内存的 50%,留足 PageCache
预防
①dmesg -T | grep -i oom 纳入每日巡检;②内存使用率 85% 告警
1 2 3 dmesg -T | grep -i "killed process" | tail -5 journalctl -k --since "24 hours ago" | grep -i oom
案例 10:auto.create.topics.enable=true 造成元数据污染
项
内容
现象
某次故障后发现集群里有 3000+ 个”莫名其妙”的 Topic,如 test、a、order_(拼写错误的)
根因
生产环境未关闭 auto.create.topics.enable,任何生产者向不存在的 Topic 发消息都会自动创建 1 分区 1 副本的 Topic(默认配置),且单副本无容灾
止血
清理这些 Topic(先确认无消费者)
根治
①auto.create.topics.enable=false;②所有 Topic 通过平台化申请创建;③将 Topic 创建/删除权限限制给运维角色
预防
①巡检脚本检查 Topic 数量突增;②检查是否存在 ReplicationFactor=1 的 Topic(必须为 0)
1 2 3 kafka-topics.sh --bootstrap-server kafka1:9092 --describe | \ grep -oP "ReplicationFactor: \K\d+" | sort | uniq -c
案例 11:误删 Topic 导致数据不可恢复
项
内容
现象
运维人员在排查问题时误执行了 kafka-topics.sh --delete,删除了核心业务 Topic
根因
①执行账号拥有过宽的删除权限;②操作前无二次确认;③无操作审计与可重放方案
止血
无法恢复!只能从上游重放数据(前提是 binlog 还在)
根治
①删除权限仅授予受控运维角色;②删除走平台审批流程;③关键操作接入堡垒机 + 双人复核
预防
①保留 delete.topic.enable=true 以支持受控清理;②平台层校验 Topic 名、影响面和备份/重放条件;③命令行保护只能作为补充
1 2 3 4 5 6 7 8 9 kafka-topics () { if [[ "$*" == *"--delete" * ]]; then echo -e "\033[31m⚠️ 危险操作!即将删除 Topic:$*\033[0m" read -p "请输入 YES 确认: " ans [ "$ans " != "YES" ] && echo "已取消" && return 1 fi command /opt/kafka/bin/kafka-topics.sh "$@ " }
案例 12:消费者位点被误重置导致数据重复处理
项
内容
现象
下游数据库出现大量重复订单,业务侧数据错乱
根因
运维为”解决积压”执行了 --reset-offsets --to-earliest,消费者从头消费全部历史数据,而消费逻辑无幂等
止血
停止消费者,清理下游脏数据
根治
①消费端实现幂等(唯一键去重);②重置 offset 前必须评估影响并拉业务方确认;③优先用 --to-latest 跳过积压
预防
①把 --reset-offsets 列入高危命令清单;②重置前先 --export 备份当前位点
1 2 3 4 5 6 7 kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group g1 \ --topic order-event --reset-offsets --to-datetime 2026-09-01T00:00:00.000 \ --export > /tmp/offset-plan.csv kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group g1 \ --topic order-event --reset-offsets --from-file /tmp/offset-plan.csv --execute
案例 13:GC 毛刺导致生产延迟周期性尖刺
项
内容
现象
生产 P99 每 30 秒出现一次 500ms 尖刺,P50 正常
根因
使用 Parallel GC,Young GC 停顿 400ms,每 30 秒一次
止血
无
根治
切换到 G1GC,MaxGCPauseMillis=200,P99 尖刺消失
预防
①新集群统一使用 G1GC;②监控 GC 停顿(见 9.6)
案例 14:数据目录拷贝导致集群 ID 冲突
项
内容
现象
新扩容的 Broker 启动报 InconsistentClusterIdException: Cluster ID mismatch
根因
新节点误挂载了其他集群的数据盘,或把其他集群的 Broker 数据目录复制过来,meta.properties 中的 cluster.id 不匹配
止血
停止 Broker,核对 ZooKeeper /cluster/id 与所有数据目录的 meta.properties,恢复正确挂载;只有确认目录不含任何需保留副本时才重新初始化为空目录
根治
①新扩容节点使用经过校验的空目录启动;②通过配置管理创建目录与权限,不复制 Broker 数据目录
预防
①把”数据目录必须为空”写进扩容 SOP;②启动前校验
1 2 3 4 5 6 DIRS="/data1/kafka /data2/kafka" for d in $DIRS ; do n=$(ls -A "$d " 2>/dev/null | wc -l) [ "$n " -ne 0 ] && echo "❌ $d 非空($n 个文件),请清空后再启动!" done
案例 15:客户端版本过老导致升级后不兼容
项
内容
现象
服务端升级到 3.7 后,一个老应用(kafka-clients 0.10.x)无法生产消息
根因
老客户端不支持新的 API 版本(客户端正常支持 2.1+,0.10 太老)
止血
该应用回滚客户端或临时走旧集群
根治
升级服务端前,先普查所有客户端的版本
预防
①建立客户端版本台账;②服务端升级前要求客户端 ≥ 2.1
事故复盘模板(每次事故后必须填写) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 ## 事故复盘报告 **事故名称** :**发生时间** :YYYY-MM-DD HH:MM **恢复时间** :YYYY-MM-DD HH:MM**影响时长** :X 分钟 **影响业务** :**影响数据** :(是否有数据丢失?多少条?)### 时间线 | 时间 | 事件 | |---|---| | HH:MM | 首次告警 | | HH:MM | 开始处置 | | HH:MM | 止血完成 | | HH:MM | 完全恢复 |### 根本原因 (技术根因 + 管理根因)### 处置过程 (做了什么,效果如何)### 改进措施 | 措施 | 负责人 | 完成时间 | 验证方式 | |---|---|---|---| | | | | |### 经验总结 (哪些做得好?哪些可以更快?)
16.1 事故预防的”十条军规”
所有 Topic 必须有明确的保留策略 (retention.ms + retention.bytes)。
生产环境:副本数 ≥ 2(推荐 3),min.insync.replicas ≥ 2,acks=all,unclean.leader.election.enable=false。
auto.create.topics.enable=false;Topic 删除保持可用,但必须通过最小 ACL、审批和审计受控执行。
任何迁移/重分配必须限速,且必须清除限速。
升级前必须锁定 inter.broker.protocol.version。
UnderReplicatedPartitions 是最高优先级指标,长期必须为 0。
Kafka 不与其他重 IO/内存型服务混部。
元数据每日备份,且每季度演练恢复。
高危命令(delete / reset-offsets / reassign / kill)纳入堡垒机审计 + 二次确认。
每次变更后 30 分钟内必须有人盯监控。
附录 A 参数速查表 A.1 Broker 端核心参数(server.properties)
参数
默认值
生产建议
说明
broker.id
0
全局唯一、递增
节点唯一标识
listeners
PLAINTEXT://:9092
显式指定协议与地址
监听地址
advertised.listeners
同 listeners
必须显式指定
对外公告地址,写错客户端连不上
log.dirs
/tmp/kafka-logs
独立数据盘,多目录
数据目录,多盘分摊
zookeeper.connect
localhost:2181
zk1:2181,zk2:2181,zk3:2181/kafka
ZK 连接串,带 chroot
num.partitions
1
按业务指定
新建 Topic 默认分区数(别依赖它)
default.replication.factor
1
3
默认副本数
min.insync.replicas
1
2
与 acks=all 配合
unclean.leader.election.enable
false
false
禁止非 ISR 副本当选 Leader
auto.create.topics.enable
true
false
禁止自动建 Topic
delete.topic.enable
true
true + 最小 ACL/审批
允许受控删除;关闭后会妨碍正常清理流程
offsets.topic.replication.factor
3
3
__consumer_offsets 副本数
offsets.topic.num.partitions
50
部署前按规模规划
内部 Topic 创建后不能靠修改此参数改变分区数
transaction.state.log.replication.factor
3
3
事务 Topic 副本数
log.retention.hours
168
按业务
数据保留时长(7 天)
log.retention.bytes
-1
按容量设置
单分区大小上限
log.segment.bytes
1073741824
1GB
日志段大小
log.retention.check.interval.ms
300000
300000
清理检查周期
log.cleanup.policy
delete
delete / compact
清理策略
log.flush.interval.messages
9223372036854775807
保持默认
依赖副本而非 fsync
log.preallocate
false
SSD 用 false
预分配文件
num.network.threads
3
核数的一半以上
网络线程
num.io.threads
8
核数 ~ 2×核数
IO 线程(最常调)
num.replica.fetchers
1
核数的一半
副本拉取线程(必调)
replica.lag.time.max.ms
30000
30000~45000
ISR 判定阈值
replica.fetch.max.bytes
1048576
≥ message.max.bytes
副本单次拉取上限
message.max.bytes
1048588
按业务
Broker 接受的 Record Batch 上限
socket.send.buffer.bytes
102400
1048576
发送缓冲
socket.receive.buffer.bytes
102400
1048576
接收缓冲
socket.request.max.bytes
104857600
保持默认
单请求上限
queued.max.requests
500
500~1000
请求队列长度
compression.type
producer
producer / lz4
Broker 端重压缩
auto.leader.rebalance.enable
true
true(或关闭后定期手动)
自动 Leader 均衡
leader.imbalance.check.interval.seconds
300
300
均衡检查间隔
zookeeper.session.timeout.ms
18000
18000~30000
ZK 会话超时(≤ ZK maxSessionTimeout)
zookeeper.connection.timeout.ms
null
通常与 session timeout 一致
null 时继承 zookeeper.session.timeout.ms
background.threads
10
10
后台清理等线程
num.recovery.threads.per.data.dir
1
2~4
启动时恢复线程
controlled.shutdown.enable
true
true
停机前迁移 Leader
controlled.shutdown.max.retries
3
3
迁移重试次数
A.2 Topic 级参数(创建时指定或通过 kafka-configs.sh 修改)
参数
默认值
说明
cleanup.policy
delete
delete / compact / compact,delete
retention.ms
604800000
7 天;-1 表示永久
retention.bytes
-1
分区大小上限
max.message.bytes
1048588
Record Batch 上限(覆盖 Broker)
segment.bytes
1073741824
段大小
segment.ms
604800000
段滚动时间
min.insync.replicas
1
最小同步副本(覆盖 Broker;生产常设 2)
min.cleanable.dirty.ratio
0.5
compact 触发水位
compression.type
producer
压缩方式
message.timestamp.type
CreateTime
CreateTime / LogAppendTime
message.timestamp.difference.max.ms
9223372036854775807
时间戳偏差容忍
unclean.leader.election.enable
false
分区级 unclean 开关
file.delete.delay.ms
60000
删除文件的延迟(防句柄泄漏)
1 2 3 4 kafka-configs.sh --bootstrap-server kafka1:9092 --alter --entity-type topics \ --entity-name order-event --add-config \ 'retention.ms=604800000,retention.bytes=107374182400,min.insync.replicas=2'
A.3 生产者参数(客户端)
参数
默认值
吞吐优先
延迟优先
可靠性优先
acks
all(3.0+)
1
1
all
linger.ms
0
50~200
0~5
5~20
batch.size
16384
1048576
16384
65536
compression.type
none
lz4/zstd
none/lz4
lz4
buffer.memory
33554432
268435456
33554432
67108864
retries
2147483647
2147483647
2147483647
2147483647
enable.idempotence
true(3.0+)
true
true
true
max.in.flight.requests.per.connection
5
5
1~5
5
request.timeout.ms
30000
30000
30000
30000
delivery.timeout.ms
120000
120000
60000
120000
max.request.size
1048576
10485760
1048576
1048576
A.4 消费者参数(客户端)
参数
默认值
说明
group.id
无
消费组 ID(必须)
auto.offset.reset
latest
latest / earliest / none
enable.auto.commit
true
生产建议 false (手动提交)
auto.commit.interval.ms
5000
自动提交间隔
session.timeout.ms
45000
心跳超时(需 ≥ Broker group.min.session.timeout.ms)
heartbeat.interval.ms
3000
心跳间隔(≤ session/3)
max.poll.interval.ms
300000
两次 poll 最大间隔,最易踩坑
max.poll.records
500
单批条数
fetch.min.bytes
1
攒批最小字节
fetch.max.wait.ms
500
攒批最大等待
fetch.max.bytes
52428800
单次 fetch 上限
max.partition.fetch.bytes
1048576
单分区 fetch 上限
isolation.level
read_uncommitted
事务消费用 read_committed
partition.assignment.strategy
Range, CooperativeSticky
建议用 CooperativeSticky(减少 STW)
A.5 ZooKeeper 参数(zoo.cfg)
参数
默认值
生产建议
说明
tickTime
2000
2000
基本时间单元
initLimit
10
10
follower 初始同步上限(×tickTime)
syncLimit
5
5
同步上限(×tickTime)
dataDir
/tmp/zookeeper
独立盘
快照目录
dataLogDir
同 dataDir
必须独立盘
事务日志目录(性能关键)
clientPort
2181
2181
客户端端口
maxClientCnxns
60
200
单 IP 最大连接数
autopurge.snapRetainCount
3
10
保留快照数
autopurge.purgeInterval
0(关闭)
24
清理间隔(小时)
maxSessionTimeout
40000(20×tickTime)
≥ Kafka session.timeout
最易踩坑
4lw.commands.whitelist
srvr
按需放行
四字命令白名单
admin.enableServer
true
false
AdminServer(3.5+ 可关)
附录 B 运维检查清单 B.1 每日巡检(10 分钟) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 #!/bin/bash BS="kafka1:9092" K="/opt/kafka/bin" RED='\033[31m' ; GREEN='\033[32m' ; NC='\033[0m' chk () { [ "$2 " -eq 0 ] && echo -e "${GREEN} ✔ $1${NC} " || echo -e "${RED} ✘ $1 : $2${NC} " ; }echo "=========== Kafka 每日巡检 $(date '+%F %T') ===========" N=$($K /kafka-broker-api-versions.sh --bootstrap-server $BS 2>/dev/null | grep -c '(id:' )echo "Broker 在线数: ${N} " UR=$($K /kafka-topics.sh --bootstrap-server $BS --describe \ --under-replicated-partitions 2>/dev/null | grep -c "Partition:" ) chk "副本不足分区数为 0" "$UR " OP=$($K /kafka-topics.sh --bootstrap-server $BS --describe \ --unavailable-partitions 2>/dev/null | grep -c "Partition:" ) chk "不可用分区数为 0" "$OP " RF1=$($K /kafka-topics.sh --bootstrap-server $BS --describe 2>/dev/null | \ grep -oP "ReplicationFactor: \K1\b" | wc -l) chk "单副本 Topic 数为 0" "$RF1 " for h in kafka1 kafka2 kafka3; do U=$(ssh $h "df /data1 | tail -1 | awk '{print \$5}' | tr -d '%'" 2>/dev/null) [ "${U:-0} " -lt 70 ] && echo -e "${GREEN} ✔ $h 磁盘 ${U} %${NC} " \ || echo -e "${RED} ✘ $h 磁盘 ${U} % (超过 70%)${NC} " done for b in 1 2 3; do R=$($K /kafka-configs.sh --bootstrap-server $BS --describe --broker $b 2>/dev/null | grep -i throttl) [ -z "$R " ] && echo -e "${GREEN} ✔ broker $b 无限速残留${NC} " \ || echo -e "${RED} ✘ broker $b 存在限速: $R${NC} " done CTRL=$($K /zookeeper-shell.sh zk1:2181/kafka get /controller 2>/dev/null | \ grep -o '"brokerid":[0-9]*' )echo "当前 Controller: $CTRL (与昨日对比,变化即为异常)" for h in zk1 zk2 zk3; do M=$(echo "srvr" | nc -w 2 $h 2181 | grep -oP "Mode: \K\w+" ) L=$(echo "mntr" | nc -w 2 $h 2181 | grep -oP "zk_avg_latency\s+\K\d+" | head -1) echo " $h : Mode=$M , avgLatency=${L} ms" done OOM=$(dmesg -T 2>/dev/null | grep -ci "killed process" ) chk "近期无 OOM 杀进程记录" "$([ "$OOM " -eq 0 ] && echo 0 || echo 1) " echo "--- 高积压消费组(LAG > 1,000,000)---" $K /kafka-consumer-groups.sh --bootstrap-server $BS --all-groups --describe 2>/dev/null | \ awk 'NR==1 || ($6+0) > 1000000 {print}' | head -20echo "=========== 巡检结束 ==========="
B.2 每周巡检
B.3 每月巡检
B.4 每季度巡检
B.5 变更检查清单(每次变更前对照) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 变更前: □ 变更内容、影响面、回滚方案明确 □ 已通知受影响业务方 □ 集群健康(UnderReplicatedPartitions = 0 ) □ 元数据已备份 □ 已确认无正在进行的重分配/迁移 □ 处于业务低峰期 变更中: □ 逐台操作,不并行 □ 每步验证(Broker 数、UR、UR 分区数) □ 有人持续盯监控(至少 30 分钟) □ 达到回滚条件立即回滚 变更后: □ ISR 全部恢复(UR = 0 ) □ 限速已清除 □ 业务方确认功能正常 □ 变更记录归档 □ 观察 24 小时
附录 C 常用脚本集合 C.1 消费积压监控与告警 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 #!/bin/bash BS="kafka1:9092" THRESHOLD=${1:-1000000} WEBHOOK="" kafka-consumer-groups.sh --bootstrap-server $BS --all-groups --describe 2>/dev/null | \ awk -v th="$THRESHOLD " ' NR > 1 && $6 ~ /^[0-9]+$/ && $6 > th { printf "⚠️ 消费积压告警 | 组: %s | Topic: %s | 分区: %s | LAG: %s\n", $1, $2, $3, $6 } ' | tee /tmp/lag-alert.txtif [ -s /tmp/lag-alert.txt ] && [ -n "$WEBHOOK " ]; then MSG=$(head -20 /tmp/lag-alert.txt | sed 's/"/\\"/g' | tr '\n' ';' ) curl -s -X POST "$WEBHOOK " \ -H 'Content-Type: application/json' \ -d "{\"msgtype\":\"text\",\"text\":{\"content\":\"$MSG \"}}" > /dev/nullfi
1 2 */5 * * * * /opt/kafka/bin/lag-monitor.sh 1000000 >> /var/log/kafka-lag.log 2>&1
C.2 Topic 一键创建(符合规范) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 #!/bin/bash set -euo pipefail TOPIC=${1:?"用法: $0 <topic> <partitions> <retention_days> [rf]"} PART=${2:?} DAYS=${3:?} RF=${4:-3} BS="kafka1:9092,kafka2:9092,kafka3:9092" RET_MS=$(( DAYS * 86400000 ))if ! [[ "$TOPIC " =~ ^[a-z][a-z0-9]*(\.[a-z0-9]+)+$ ]]; then echo "❌ Topic 命名不符合规范(要求 小写.点分,如 order.event.v1): $TOPIC " exit 1fi [ "$RF " -lt 2 ] && echo "❌ 生产环境副本数必须 ≥ 2" && exit 1echo "创建 Topic: $TOPIC " echo " 分区数: $PART 副本数: $RF 保留: ${DAYS} 天" /opt/kafka/bin/kafka-topics.sh --bootstrap-server $BS \ --create --topic "$TOPIC " \ --partitions "$PART " --replication-factor "$RF " \ --config "retention.ms=$RET_MS " \ --config "min.insync.replicas=2" \ --config "compression.type=lz4" \ --config "max.message.bytes=1048576" /opt/kafka/bin/kafka-topics.sh --bootstrap-server $BS --describe --topic "$TOPIC "
C.3 分区迁移(带限速 + 自动清限速) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 #!/bin/bash set -euo pipefail JSON=${1:?"用法: $0 <reassign.json> [throttle]"} THROTTLE=${2:-104857600} BS="kafka1:9092" K="/opt/kafka/bin" trap 'echo "脚本被中断;限速仍保留。恢复后重新运行本脚本或执行 --verify。" >&2; exit 130' INT TERMecho "[$(date '+%T') ] 开始重分配,限速 $((THROTTLE/1048576) ) MB/s" $K /kafka-reassign-partitions.sh --bootstrap-server $BS \ --reassignment-json-file "$JSON " --execute --throttle "$THROTTLE " echo "[$(date '+%T') ] 等待完成..." while true ; do OUT=$($K /kafka-reassign-partitions.sh --bootstrap-server $BS \ --reassignment-json-file "$JSON " --verify 2>&1) echo "$OUT " | grep -c "still in progress" | \ xargs -I{} echo "[$(date '+%T') ] 进行中分区数: {}" echo "$OUT " | grep -q "still in progress" || break sleep 60done echo "[$(date '+%T') ] 重分配完成;最终 --verify 已清除本次重分配限速"
--verify 只在确认重分配完成后清除相关限速。不要用 trap ... EXIT 无条件删限速,否则终端断开或脚本报错时,后台迁移仍可能继续并打满网络。
C.4 集群健康一键体检 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 #!/bin/bash BS="kafka1:9092" K="/opt/kafka/bin" echo "╔══════════════════════════════════════════════╗" echo "║ Kafka 集群健康体检 $(date '+%F %T') ║" echo "╚══════════════════════════════════════════════╝" echo ; echo "【1】Broker 列表" $K /kafka-broker-api-versions.sh --bootstrap-server $BS 2>/dev/null | \ grep '(id:' | awk -F'(' '{print " " $1}' echo ; echo "【2】副本不足分区" UR=$($K /kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions 2>/dev/null) [ -z "$UR " ] && echo " ✅ 无(健康)" || echo "$UR " | sed 's/^/ /' echo ; echo "【3】不可用分区" OP=$($K /kafka-topics.sh --bootstrap-server $BS --describe --unavailable-partitions 2>/dev/null) [ -z "$OP " ] && echo " ✅ 无(健康)" || echo "$OP " | sed 's/^/ /' echo ; echo "【4】Topic 统计" TOTAL=$($K /kafka-topics.sh --bootstrap-server $BS --list 2>/dev/null | grep -vc "^__" ) PARTS=$($K /kafka-topics.sh --bootstrap-server $BS --describe 2>/dev/null | grep -c "Partition:" )echo " 业务 Topic 数: $TOTAL " echo " 分区总数: $PARTS " echo ; echo "【5】Controller" echo " $(/opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controller 2>/dev/null | grep 'brokerid') " echo " epoch: $(/opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controller_epoch 2>/dev/null | tail -1) " echo ; echo "【6】消费组总览" $K /kafka-consumer-groups.sh --bootstrap-server $BS --list 2>/dev/null | \ grep -v "^$" | wc -l | xargs -I{} echo " 消费者组数量: {}" echo ; echo "【7】Top 10 积压" $K /kafka-consumer-groups.sh --bootstrap-server $BS --all-groups --describe 2>/dev/null | \ awk 'NR>1 && $6 ~ /^[0-9]+$/ {print $6, $1, $2, $3}' | \ sort -rn | head -10 | awk '{printf " LAG=%-12s %s / %s / %s\n",$1,$2,$3,$4}' echo ; echo "【8】ZooKeeper" for h in zk1 zk2 zk3; do R=$(echo "srvr" | nc -w 2 $h 2181 2>/dev/null | grep -E "Mode|Node count" | tr '\n' ' ' ) echo " $h : ${R:-不可达} " done echo ; echo "体检完成。"
C.5 日志清理与归档 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 #!/bin/bash LOG_DIR=/opt/kafka/logs KEEP_DAYS=7echo "[$(date '+%F %T') ] 清理 Kafka 日志(保留 $KEEP_DAYS 天)" BEFORE=$(du -sh $LOG_DIR | cut -f1) find $LOG_DIR -name "*.log.*" -mtime +$KEEP_DAYS -delete find $LOG_DIR -name "*.out.*" -mtime +$KEEP_DAYS -delete find /tmp -name "jstack-*" -mtime +$KEEP_DAYS -delete 2>/dev/null find /tmp -name "kafka-logs-*" -mtime +$KEEP_DAYS -delete 2>/dev/null AFTER=$(du -sh $LOG_DIR | cut -f1)echo " 清理前: $BEFORE 清理后: $AFTER "
C.6 systemd 服务文件(生产模板) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 [Unit] Description =Apache Kafka BrokerDocumentation =https://kafka.apache.org/documentation/Requires =network.target remote-fs.targetAfter =network.target remote-fs.target zookeeper.serviceWants =zookeeper.service[Service] Type =simpleUser =kafkaGroup =kafkaEnvironment ="JAVA_HOME=/usr/lib/jvm/java-11-openjdk" Environment ="KAFKA_HEAP_OPTS=-Xms16g -Xmx16g" Environment ="KAFKA_JVM_PERFORMANCE_OPTS=-server -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ParallelRefProcEnabled -XX:+DisableExplicitGC -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/opt/kafka/logs -Xlog:gc*:file=/opt/kafka/logs/gc.log:time,uptime,level,tags:filecount=10,filesize=100M" Environment ="KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf" Environment ="LOG_DIR=/opt/kafka/logs" ExecStart =/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.propertiesExecStop =/bin/kill -TERM $MAINPID Restart =on -failureRestartSec =10 TimeoutStopSec =180 LimitNOFILE =1000000 LimitNPROC =65536 OOMScoreAdjust =-500 [Install] WantedBy =multi-user.target
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 [Unit] Description =Apache ZooKeeperRequires =network.targetAfter =network.target[Service] Type =forkingUser =zookeeperGroup =zookeeperEnvironment ="JAVA_HOME=/usr/lib/jvm/java-11-openjdk" Environment ="SERVER_JVMFLAGS=-Xms4g -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/zookeeper -Xlog:gc*:file=/var/log/zookeeper/gc.log:time:filecount=5,filesize=50M" ExecStart =/opt/zookeeper/bin/zkServer.sh startExecStop =/opt/zookeeper/bin/zkServer.sh stopRestart =on -failureRestartSec =10 LimitNOFILE =65536 [Install] WantedBy =multi-user.target
C.7 常用 Shell 别名(提升效率) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 export KAFKA_HOME=/opt/kafkaexport PATH=$KAFKA_HOME /bin:$PATH alias kt='kafka-topics.sh --bootstrap-server kafka1:9092' alias kcl='kafka-consumer-groups.sh --bootstrap-server kafka1:9092' alias kl='kafka-topics.sh --bootstrap-server kafka1:9092 --list | grep -v "^__"' alias kur='kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions' alias kdesc='kafka-topics.sh --bootstrap-server kafka1:9092 --describe' alias kbrokers='kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep "(id:"' alias kctrl='/opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controller' alias kzk='echo "mntr" | nc -w 3 zk1 2181' klag () { [ -z "$1 " ] && echo "用法: klag <topic>" && return 1 kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --all-groups --describe | \ awk -v t="$1 " 'NR==1 || $2==t' }kdist () { [ -z "$1 " ] && echo "用法: kdist <topic>" && return 1 kafka-topics.sh --bootstrap-server kafka1:9092 --describe --topic "$1 " | \ awk '/Partition:/ {for(i=1;i<=NF;i++) if($i=="Leader:") print $(i+1)}' | \ sort | uniq -c | sort -rn }
C.8 JMX 快速查询(jmxterm) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 cat > /opt/kafka/bin/jmx-query.txt <<'EOF' domain kafka.server bean kafka.server:type =ReplicaManager,name=UnderReplicatedPartitions get Value bean kafka.server:type =ReplicaManager,name=PartitionCount get Value bean kafka.server:type =ReplicaManager,name=LeaderCount get Value bean kafka.controller:type =KafkaController,name=ActiveControllerCount get Value bean kafka.network:type =RequestChannel,name=RequestQueueSize get Value bean kafka.server:type =BrokerTopicMetrics,name=MessagesInPerSec get Count bean kafka.server:type =BrokerTopicMetrics,name=BytesInPerSec get Count bean kafka.server:type =BrokerTopicMetrics,name=BytesOutPerSec get Count bean kafka.log:type =LogFlushStats,name=LogFlushRateAndTimeMs get Count bean kafka.server:type =KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent get OneMinuteRate EOF java -jar /opt/jmxterm/jmxterm-1.0.4-uber.jar -l localhost:9999 -i /opt/kafka/bin/jmx-query.txt
C.9 关键 JMX 指标清单(监控必采集)
JMX MBean
指标
含义
告警阈值
kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
Value
副本不足分区数
> 0 立即告警
kafka.server:type=ReplicaManager,name=OfflinePartitionsCount
Value
不可用分区数
> 0 严重
kafka.controller:type=KafkaController,name=ActiveControllerCount
Value
本节点是否为 Controller
全集群和应为 1
kafka.server:type=ReplicaManager,name=IsrShrinksPerSec
Count
ISR 收缩次数
持续增长告警
kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec
Count/OneMinuteRate
消息写入速率
业务基线浮动 ±50%
kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec
OneMinuteRate
入流量
接近带宽上限
kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec
OneMinuteRate
出流量
接近带宽上限
kafka.network:type=RequestChannel,name=RequestQueueSize
Value
请求队列长度
> 100 持续 1 分钟
kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent
OneMinuteRate
IO 线程空闲率
< 30%
kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce
99thPercentile
生产 P99
> 500ms
kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchConsumer
99thPercentile
消费 P99
> 500ms
kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs
99thPercentile
刷盘耗时
> 100ms
kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica
Value
最大副本延迟
> 10000
kafka.server:type=FetchQueueSizeMetrics,name=FetchQueueSize(或 follower 队列)
Value
副本拉取队列
持续 > 0
kafka.coordinator.group:type=GroupMetadataManager,name=NumGroups
Value
消费者组数
突增
java.lang:type=GarbageCollector,name=G1 Young Generation
CollectionTime
GC 时间
单次 > 3s
附录 D 学习路径与资料 D.1 从入门到精通的学习路径 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 【第 1 周】跑起来 + 建立感性认识 · 第 0 章:单机跑通,理解生产者/消费者/Topic · 第 1 章:背下核心概念表,理解 ISR/LEO/HW · 会用:kafka-topics / kafka-console-producer / kafka-console-consumer 【第 2-3 周】搭一套真实集群 · 第 2-4 章:3 台 ZK + 3 台 Kafka,按生产参数配置 · 第 11 章:加上 SASL/SSL/ACL(一定在测试环境先做) · 第 6 章:把命令行工具全部敲一遍 【第 4-6 周】监控与调优 · 第 8 章:搭 Prometheus + Grafana + 告警 · 第 9 章:做一次压测,找到自己集群的瓶颈 · 第 7 章:把日常运维操作写成脚本 【第 7-8 周】容灾与故障 · 第 12-14 章:扩容、升级、容灾演练 · 第 10 章:对照故障速查表,逐个理解根因 · 第 16 章:把 15 个案例对照自己的集群做检查 【长期】形成自己的知识体系 · 读 Kafka 源码关键路径:Log.append / ReplicaManager / GroupCoordinator · 关注 Kafka KIP 提案,理解版本演进方向 · 参与社区,看别人踩的坑
D.2 官方与权威资料
D.3 生产环境检查的”终极十问” 在集群上线前,用这十个问题自查:
副本数是 3 吗?min.insync.replicas 是 2 吗?acks 是 all 吗?unclean.leader.election.enable 是 false 吗?
每个 Topic 都有明确的 retention.ms 和 retention.bytes 吗?
auto.create.topics.enable 关了吗?Topic 删除是否已通过最小 ACL、审批和审计受控?
ZK 是 3 或 5 节点、独立部署、事务日志在独立盘吗?
Kafka 与 ZK 是否混部?Kafka 是否与其他重 IO 服务混部?
UnderReplicatedPartitions 有监控和告警吗?告警能到人吗?
磁盘 70%/85% 有分级告警吗?容量能撑多久算过吗?
元数据备份开启了吗?演练过恢复吗 ?
高危命令(delete/reset-offsets/reassign)有审计和二次确认吗?
上一次故障演练是什么时候?演练结果有记录吗?
如果这十个问题里有任何一个答案是”没有”,那么这个集群就还没到”生产就绪”。
结语 Kafka 的运维并不复杂,难的是细节的持续执行 。
这套手册里 90% 的故障,根因都可以归纳为三类:
配置不当 —— 副本数、acks、min.insync.replicas、保留策略,这四个参数决定了 80% 的稳定性;
变更失控 —— 迁移不限速、升级不锁协议版本、限速不清除、高危命令无审计;
监控缺失 —— UnderReplicatedPartitions、磁盘水位、消费积压、GC 停顿,这四个指标能提前发现绝大多数问题。
真正的运维能力,不是”出事之后能修好”,而是”让事不发生”。
最后三条建议:
把所有手工操作脚本化 ,脚本化的操作才能被 review、被复用、被新人执行;
把所有假设验证化 ,”我以为没问题”是事故最常见的开场白;
把每次故障文档化 ,本手册第 16 章的每一个案例,都是某个人某天凌晨的真实经历。
版本说明:本手册基于 Apache Kafka 3.0.x ~ 3.9.x(ZooKeeper 模式)编写。 Kafka 4.0 已移除 ZooKeeper 模式。新集群使用 KRaft,存量 ZooKeeper 集群应在 3.x 阶段制定并演练迁移计划(见 13.5)。