Kafka 3.x 实战运维手册(ZooKeeper 版,示例基线 3.6.2)

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
# 1. 准备 JDK(Kafka 3.6.2 支持 JDK 8/11/17;JDK 8 已弃用,建议 11 或 17)
java -version

# 2. 下载并解压(本文固定以 3.6.2 为示例基线,不代表当前最新版本)
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.tgz
ln -s /opt/kafka_2.13-3.6.2 /opt/kafka

# 3. 启动自带的 ZooKeeper(仅供体验!生产必须独立部署)
cd /opt/kafka
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

# 4. 启动 Kafka
bin/kafka-server-start.sh -daemon config/server.properties

# 5. 创建主题
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--create --topic quickstart --partitions 3 --replication-factor 1

# 6. 生产消息
echo -e "hello\nkafka\nworld" | bin/kafka-console-producer.sh \
--bootstrap-server localhost:9092 --topic quickstart

# 7. 消费消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic quickstart --from-beginning --timeout-ms 5000

# 8. 关闭
bin/kafka-server-stop.sh
bin/zookeeper-server-stop.sh

0.2 你该知道的 5 件事

  1. Kafka 3.x 仍支持 ZooKeeper,Kafka 4.0 已移除 ZooKeeper 模式。3.x 是 ZooKeeper 模式的最后一个大版本,存量集群应规划向 KRaft 迁移。
  2. Kafka 的吞吐来自顺序写、页缓存、批量与零拷贝等机制。磁盘吞吐和尾延迟仍会直接影响生产请求、副本同步与恢复速度,不能忽略。
  3. 分区数一旦增加不可减少。Kafka 3.x 不支持减少 Topic 分区数。
  4. 副本数(replication.factor)决定了容灾能力,1 副本 = 无容灾,生产最低 2,推荐 3。
  5. 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 所在的 Broker
3. Leader 写入本地 Log(PageCache → 磁盘)
4. Follower 主动拉取(fetch)同步
5. 当 ISR 中所有副本都写入成功(acks=all)→ Leader 更新 HW
6. 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(不要超过 12GB,否则 GC 停顿不可控)
剩余内存全部留给 OS PageCache(Kafka 主要靠 PageCache 缓存热数据)
即:32GB 内存 → Heap 8GB + PageCache ~20GB

⚠️ 常见误区:给 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.dir 分摊,Kafka 会自动把分区打散到各目录
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
# /etc/security/limits.conf 追加
* soft nofile 1000000
* hard nofile 1000000
root soft nofile 1000000
root hard nofile 1000000

# /etc/security/limits.d/20-nproc.conf 确认 nproc 不要过小
* soft nproc 65535

# 验证(重新登录后)
ulimit -n

若用 systemd 管理,还需在 unit 文件加 LimitNOFILE=1000000。

2.2.2 虚拟内存 / Swap

1
2
3
4
5
6
7
# 1. 关闭 swap(Kafka 通常不希望被换出,但保留少量 swap 防 OOM 也可)
swapoff -a
# 永久生效:注释 /etc/fstab 中的 swap 行

# 2. 调整 swappiness(若不关闭 swap,建议设为 1)
sysctl -w vm.swappiness=1
echo '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
# 每个 socket 发送缓冲
net.core.wmem_default = 262144
net.core.rmem_default = 262144
net.core.wmem_max = 2097152
net.core.rmem_max = 2097152
# TCP 缓冲区自动调整
net.ipv4.tcp_rmem = 4096 65536 16777216
net.ipv4.tcp_wmem = 4096 65536 16777216
# TIME_WAIT 快速回收与重用
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
# TCP 连接超时
net.ipv4.tcp_fin_timeout = 30
# MTU 探测
net.ipv4.tcp_mtu_probing = 1
EOF

sysctl -p

2.2.4 文件系统与挂载

1
2
3
4
5
6
7
# XFS 与 ext4 都可用于 Kafka;以下以 XFS 为例,最终应按实际磁盘和内核压测
mkfs.xfs -f /dev/nvme0n1
# 挂载参数:noatime 减少元数据写入
mount -o noatime,nodiratime /dev/nvme0n1 /data1

# /etc/fstab 持久化
/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
# 使用 chrony
yum install -y chrony
cat >> /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/enabled
echo never > /sys/kernel/mm/transparent_hugepage/defrag

# 永久生效(写入 /etc/rc.local 或 systemd service)

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.gz
ln -s /opt/apache-zookeeper-3.8.4-bin /opt/zk

# 创建数据目录
mkdir -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
# 心跳周期(tick 为单位,2000ms)
tickTime=2000
# Leader 与 Follower 初始连接超时(tick 倍数)
initLimit=10
# Leader 与 Follower 数据同步超时(tick 倍数)
syncLimit=5
# 客户端端口
clientPort=2181
# 数据目录
dataDir=/data/zookeeper/data
# 事务日志目录(必须独立磁盘!事务日志是顺序写,性能敏感)
dataLogDir=/data/zookeeper/logs
# 单个客户端 IP 的最大连接数;按 Broker 数和运维连接预留余量
maxClientCnxns=200
# 自动清理:保留 3 个快照,每 24 小时清理
autopurge.snapRetainCount=3
autopurge.purgeInterval=24
# 集群成员:server.myid=host:2888(选主通信):3888(选举通信)
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
# 不使用 AdminServer 时关闭,减少额外暴露面
admin.enableServer=false
admin.serverPort=8080

设置 myid(每台机器不同):

1
2
3
4
5
6
# zk-node1
echo 1 > /data/zookeeper/data/myid
# zk-node2
echo 2 > /data/zookeeper/data/myid
# zk-node3
echo 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

# 查看角色(会显示 leader 或 follower)
/opt/zk/bin/zkServer.sh status

# 集群健康检查
echo ruok | nc 127.0.0.1 2181 # 返回 imok
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        # leader / follower
zk_followers # follower 数量(leader 上)
zk_synced_followers # 已同步 follower 数
zk_pending_syncs # 待同步数(应长期为 0)
zk_avg_latency # 平均延迟(应 < 10ms)
zk_max_latency # 最大延迟(尖刺需关注)
zk_outstanding_requests# 排队请求(长期 > 0 说明过载)
zk_num_alive_connections # 活跃连接数
zk_znode_count # 节点数(Kafka 集群大时可达数十万)
zk_watch_count # watch 数(Kafka 会产生大量 watch)
zk_ephemerals_count # 临时节点数(每个 Broker 会注册临时节点)
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
# 优先使用 ZooKeeper 自带清理脚本,它会依据 zoo.cfg 中的目录安全清理
/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
# /etc/systemd/system/zookeeper.service
[Unit]
Description=Apache ZooKeeper
After=network.target

[Service]
Type=forking
User=zookeeper
Group=zookeeper
Environment="JAVA_HOME=/usr/lib/jvm/java-11-openjdk"
ExecStart=/opt/zk/bin/zkServer.sh start
ExecStop=/opt/zk/bin/zkServer.sh stop
ExecReload=/opt/zk/bin/zkServer.sh restart
Restart=on-failure
RestartSec=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-openjdk
# ZK 堆不宜超过 8GB(否则 Full GC 停顿长)
export 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.tgz
ln -s /opt/kafka_2.13-3.6.2 /opt/kafka

# 创建数据目录(多盘建议多目录)
mkdir -p /data1/kafka /data2/kafka
mkdir -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
############################# Server Basics #############################
# 节点唯一 ID,必须全局唯一且不可重复
broker.id=1
# 是否允许删除 Topic;默认 true,生产通过 ACL 和变更流程限制 Delete 权限
delete.topic.enable=true
# 自动创建 Topic(生产必须关闭!否则拼错 Topic 名会造成元数据污染)
auto.create.topics.enable=false

############################# Socket Server Settings #############################
# 监听地址(注意:Kafka 需要在所有网卡上可达)
listeners=PLAINTEXT://0.0.0.0:9092
# 对外公告地址(生产必须显式指定,否则内网 IP 会暴露给客户端导致连不上)
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 Basics #############################
# 数据目录(多盘用逗号分隔,Kafka 自动打散分区)
log.dirs=/data1/kafka,/data2/kafka
# 新建 Topic 默认分区数(建议按业务显式指定,不要依赖默认)
num.partitions=6
# 默认副本数(生产建议 3)
default.replication.factor=3
# 是否定期把 Leader 调整回首选副本,以改善 Leader 分布
auto.leader.rebalance.enable=true

############################# Internal Topic Settings #############################
# __consumer_offsets 的副本数(生产必须 >= 3)
offsets.topic.replication.factor=3
# __consumer_offsets 分区数;内部 Topic 创建后不要再通过此参数修改
offsets.topic.num.partitions=50
# 事务日志 Topic 副本数
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
transaction.state.log.num.partitions=50

############################# Log Retention Policy #############################
# 日志保留时间(默认 7 天)
log.retention.hours=168
# 保留大小上限(-1 表示不限,建议设置防单分区撑爆磁盘)
log.retention.bytes=-1
# 日志段大小(1GB,太大会导致清理不及时,太小文件数过多)
log.segment.bytes=1073741824
# 段滚动检查间隔(检查段是否达到保留条件)
log.retention.check.interval.ms=300000
# 消息时间戳类型:CreateTime(生产者时间)或 LogAppendTime(Broker 时间)
# 强烈建议 CreateTime + 业务保证时钟同步
log.message.timestamp.type=CreateTime
# 最大时钟偏差容忍(时间戳超前/滞后超过该值会被拒绝)
log.message.timestamp.difference.max.ms=9223372036854775807
#-- 清理策略:delete(按时间/大小删除)或 compact(按 key 压缩)
log.cleanup.policy=delete

############################# Log Flush Policy #############################
# 不主动按消息数触发 fsync,依赖 PageCache、操作系统刷盘和副本机制
log.flush.interval.messages=9223372036854775807
# `log.flush.interval.ms` 的默认值是 null;不要在 properties 中写字符串 null

############################# Replication Settings #############################
# 分区的默认最小同步副本数(与 acks=all 配合使用)
min.insync.replicas=2
# 副本落后多少时间被踢出 ISR(默认 30s,不建议调小)
replica.lag.time.max.ms=30000
# 复制线程数(副本多的集群可加大)
num.replica.fetchers=4
# 副本 fetch 的最小/最大字节(加大可提升副本同步吞吐)
replica.fetch.min.bytes=1
replica.fetch.max.bytes=10485760
replica.fetch.wait.max.ms=500
# 是否允许非 ISR 副本当选 Leader(生产必须 false!)
unclean.leader.election.enable=false
# 整个副本 Fetch 响应允许返回的最大字节数
replica.fetch.response.max.bytes=10485760

############################# ZooKeeper #############################
# ZK 连接串
zookeeper.connect=10.0.0.11:2181,10.0.0.12:2181,10.0.0.13:2181
# ZK 会话超时(数据量大时需加大,默认 18000)
zookeeper.session.timeout.ms=18000
# ZK 连接超时
zookeeper.connection.timeout.ms=18000
# ZK 最大待处理请求(背压保护)
zookeeper.max.in.flight.requests=10

############################# Threads & Performance #############################
# 后台线程处理消息数
background.threads=10
# 复制限速应在重分配期间通过 kafka-reassign-partitions.sh 动态设置,
# 不应长期写入静态 server.properties

############################# Group Coordinator #############################
# 消费者组 coordinator 加载延迟
group.initial.rebalance.delay.ms=3000
# 消费者 offset 提交 topic 的压缩类型
offsets.topic.compression.codec=0

############################# Metrics #############################
# JMX 端口(监控必须!)
# 通过 KAFKA_JMX_OPTS 或 kafka-server-start.sh 的 JMX_PORT 环境变量设置

############################# Security #############################
# 安全配置见第 11 章,此处保持 PLAINTEXT

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
# 关键:把 DEBUG 关掉,TRACE 绝对不要开
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
# 请求日志(排查慢请求用,平时 INFO 即可)
log4j.logger.kafka.request.logger=WARN, requestAppender
log4j.logger.kafka.network.Processor=WARN, requestAppender
log4j.logger.kafka.server.KafkaApis=WARN, requestAppender
# 副本 Fetcher(非常吵,除非排查副本问题否则关闭)
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
# 推荐的 JVM 参数(JDK 11 + G1GC)
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=9999
export 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
# 启动(新增节点建议前台启动一次,确认无误后再 daemon)
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

# 查看日志
tail -f /var/log/kafka/server.log

# 验证进程
jps -l | grep kafka

# 验证端口
netstat -tlnp | grep 9092

# 验证集群(列出 broker)
/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
# /etc/systemd/system/kafka.service
[Unit]
Description=Apache Kafka
After=network.target zookeeper.service
Wants=zookeeper.service

[Service]
Type=simple
User=kafka
Group=kafka
Environment="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.properties
ExecStop=/bin/kill -TERM $MAINPID
Restart=on-failure
RestartSec=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
# 1. 先检查是否有分区 Leader 在本节点(有则需迁移)
/opt/kafka/bin/kafka-topics.sh --bootstrap-server 10.0.0.21:9092 \
--describe | grep "Leader: 1"

# 2. (可选)主动触发 Leader 迁移,减少停机时的不可用窗口
# 见 7.x 章 kafka-leader-election.sh

# 3. 优雅停机(systemd 向主进程发送 TERM,Kafka 会执行受控关闭)
systemctl stop kafka

# 4. 确认进程退出、ZK 临时节点消失

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
# 限制生产者:每秒最大 10MB 生产
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
# 动态修改 Broker 配置
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'

# 动态修改 Topic 配置
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:9092
export KAFKA_HOME=/opt/kafka
export 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
# 列出所有 Topic(含内部 Topic)
kafka-topics.sh --bootstrap-server $BS --list

# 列出并排除内部 Topic
kafka-topics.sh --bootstrap-server $BS --list --exclude-internal

# 创建 Topic(推荐显式指定所有参数)
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

# 查看单个 Topic 详情(分区分布、Leader、ISR)
kafka-topics.sh --bootstrap-server $BS --describe --topic order-event

# 查看所有 Topic 详情
kafka-topics.sh --bootstrap-server $BS --describe

# 只看特定状态的 Topic(如副本不足)
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

# 精确确认 Topic 是否存在
kafka-topics.sh --bootstrap-server $BS --list --topic '^order-event$'

# 增加分区(只能增不能减!)
kafka-topics.sh --bootstrap-server $BS --alter \
--topic order-event --partitions 24

# 修改 Topic 配置
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

# 删除 Topic(需服务端 delete.topic.enable=true)
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
# 控制台生产者(手动输入,Ctrl+C 退出)
kafka-console-producer.sh --bootstrap-server $BS --topic order-event

# 带 key 生产(key:value 格式)
kafka-console-producer.sh --bootstrap-server $BS --topic order-event \
--property parse.key=true --property key.separator=:

# kafka-console-producer.sh:Kafka 自带的控制台生产者,从标准输入读一行发一条消息。
# --bootstrap-server $BS:指定接入的 broker 地址(集群入口,只用于拉取元数据,不必写全部 broker)。
# --topic order-event:消息写入 order-event 这个主题。
# --property parse.key=true:告诉生产者从输入行里解析出 key。默认是 false,即整行都当 value、key 为 null。
# --property key.separator=::指定分隔符为冒号。于是输入行会按第一个冒号切成两部分,左边是 key、右边是 value。

# 从文件生产
kafka-console-producer.sh --bootstrap-server $BS --topic order-event < data.txt

# 指定 acks 与压缩
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

# 显示 key 和 offset
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

# 从指定 offset 开始消费 10 条;Kafka 3.6 的控制台消费者没有“自动取最后 n 条”选项
kafka-console-consumer.sh --bootstrap-server $BS --topic order-event \
--partition 0 --offset 12345 --max-messages 10

# 格式化输出(JSON)
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-ID
order-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

# 查看组状态(Stable / PreparingRebalance / Empty / Dead)
kafka-consumer-groups.sh --bootstrap-server $BS --describe \
--group order-consumer-group --state

# 重置 offset(⚠️ 高风险操作)
# 场景1:重置到最早
kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \
--topic order-event --reset-offsets --to-earliest --execute

# 场景2:重置到最新
kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \
--topic order-event --reset-offsets --to-latest --execute

# 场景3:重置到指定 offset
kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \
--topic order-event:0 --reset-offsets --to-offset 100000 --execute

# 场景4:按时间重置(回到 2 小时前)
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

# 场景5:位移若干条(相对回退)
kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \
--topic order-event --reset-offsets --shift-by -1000 --execute

# 场景6:导出重置计划(不执行,先看)
kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \
--topic order-event --reset-offsets --to-earliest --dry-run

# 场景7:导出到 CSV 备份
kafka-consumer-groups.sh --bootstrap-server $BS --group order-consumer-group \
--topic order-event --reset-offsets --to-current --export > offsets.csv
# 再从 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
# 第一步:生成候选分配方案
# 方式A:根据 broker 列表自动生成(保持主题均衡)
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
# 第二步:执行重分配(-throttle 限速,单位字节/秒,强烈建议加!)
kafka-reassign-partitions.sh --bootstrap-server $BS \
--reassignment-json-file reassign.json \
--throttle 104857600 --execute

# 第三步:验证进度(反复执行直到全部 completed)
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
# 若已 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-partitions

# 触发 Preferred Leader 选举(把 Leader 拉回首选位置)
cat > 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

# 或者用 --all-topic-partitions 对所有分区做 preferred 选举
kafka-leader-election.sh --bootstrap-server $BS \
--election-type preferred --all-topic-partitions

# 强制指定某分区 Leader(unclean 选举,会丢数据!慎用)
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/
# 输出:
# 00000000000000000000.log 1.0G
# 00000000000000000000.index 10M
# 00000000000000000000.timeindex 10M
# 00000000000000012345.log ...
1
2
# 查看日志段的物理信息(文件大小、offset 范围、时间范围)
kafka-log-dirs.sh --bootstrap-server $BS --topic-list order-event --describe
1
2
3
4
5
6
7
8
9
10
11
# Dump 日志文件内容(配合 kafka-dump-log.sh)
kafka-dump-log.sh --files /data1/kafka/order-event-0/00000000000000000000.log \
--print-data-log --deep-iteration | head -20

# 查看 segment 索引
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
# 查看所有 Broker 的动态配置
kafka-configs.sh --bootstrap-server $BS --describe --entity-type brokers --all

# 查看某 Broker 的配置(包含默认值)
kafka-configs.sh --bootstrap-server $BS --describe \
--entity-type brokers --entity-name 1 --all

# 查看 Topic 的所有配置(含默认)
kafka-configs.sh --bootstrap-server $BS --describe \
--entity-type topics --entity-name order-event --all

# 查看集群 API 版本(判断客户端兼容性)
kafka-broker-api-versions.sh --bootstrap-server $BS

# 查看 Broker 支持的 API 版本范围(不能据此精确判断二进制版本或集群 ID)
kafka-broker-api-versions.sh --bootstrap-server $BS | head -3

# ZooKeeper 模式查看集群 ID
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

# 输出解读:
# 1000000 records sent, 125000.0 records/sec (122.07 MB/sec),
# 245.32 ms avg latency, 890.00 ms max latency

# 消费者压测;--threads 在 Kafka 3.6 中已弃用且会被忽略
kafka-consumer-perf-test.sh --bootstrap-server $BS \
--topic perf-test \
--messages 1000000 \
--show-detailed-stats

# 端到端延迟压测:位置参数依次为 broker、topic、消息数、acks、消息大小
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
# /opt/kafka/bin/daily-check.sh —— 每日巡检脚本
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-partitions

echo "----- [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 -10

echo "----- [7] 磁盘 IO 与负载 -----"
iostat -x 1 2 | tail -20
uptime

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
# 查看当前 Controller(哪个 Broker 是 Controller)
/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
# 返回 epoch 值,每次 Controller 切换会 +1

> 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 切换(维护场景):

1
2
# 方式:重启当前 Controller 所在 Broker(会影响其上的所有 Leader)
# 更平滑的方式:先迁移其上的 Leader,再重启

7.3 Topic 生命周期管理

7.3.1 Topic 创建规范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 命名规范:{业务域}.{数据类型}.{版本}  例如:order.event.v1
# 分区数规范:根据峰值吞吐 / 单分区承载能力计算
# 单分区经验值:10~50 MB/s(取决于磁盘和网络)
# 分区数 = ceil(峰值吞吐 / 单分区能力) 再向上取整到便于均衡的数(如 6/12/24)

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 的 hash 分配改变,同一 key 可能落到新分区
# 影响:消费顺序性(如果业务依赖同 key 有序),需要评估

分区扩容的坑:

  • 若生产者按 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
# 1. 确认 delete.topic.enable=true;该参数为静态 Broker 配置,修改后需滚动重启
kafka-configs.sh --bootstrap-server $BS --describe --entity-type brokers --entity-name 1 --all | grep delete.topic

# 2. 停止所有生产/消费(关键!)
# 3. 删除
kafka-topics.sh --bootstrap-server $BS --delete --topic old-topic

# 4. 验证删除
kafka-topics.sh --bootstrap-server $BS --list | grep old-topic

# 5. 若删除长时间未完成,检查 Controller、离线 Broker、Broker 日志和目录权限
kafka-topics.sh --bootstrap-server $BS --describe --unavailable-partitions
# 消费者是否在线不会阻止 Topic 删除。不要手工删除 ZK 元数据或 Broker 数据目录;
# 这会绕过 Controller 状态机并可能造成元数据与磁盘状态不一致。

7.4 数据目录运维

1
2
3
4
5
6
# 查看各 Broker 的日志目录使用情况
kafka-log-dirs.sh --bootstrap-server $BS --describe --topic-list order.event.v1

# 输出解读:
# {"brokers":[{"broker":1,"logDirs":[{"logDir":"/data1/kafka","error":null,
# "partitions":[{"partition":"order.event.v1-0","size":1073741824,"offsetLag":0,"isFuture":false}]}]}]}

offsetLag 的含义:该副本与 Leader 的 offset 差距。正常应为 0;长期 > 0 表示副本同步有问题。

磁盘满的应急处理:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 1. 快速定位大分区
du -sh /data1/kafka/* | sort -rh | head -20

# 2. 若是日志类 Topic,临时缩短保留时间
kafka-configs.sh --bootstrap-server $BS --alter \
--entity-type topics --entity-name log-topic \
--add-config retention.ms=3600000 # 1 小时

# 3. 手动触发清理(等待 log.retention.check.interval.ms,默认 5 分钟)

# 4. 若还是不够,迁移部分分区到其他盘/节点
kafka-reassign-partitions.sh --bootstrap-server $BS \
--reassignment-json-file move.json --throttle 52428800 --execute

# 5. 若空间仍不足:扩容文件系统,或停止受影响 Broker 后按已批准的数据丢失方案处置
# 不要在 Broker 运行时手工删除 segment;“已消费”不等于 Kafka 可以安全遗忘该文件。

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
# 查看 order-event 所有分区的最新 offset
kafka-run-class.sh kafka.tools.GetOffsetShell \
--bootstrap-server 10.0.0.21:9092 --topic order-event --time -1

# 输出:
# order-event:0:1234567
# order-event:1:2234567
# ...

# 统计总消息数(最新 offset 之和 - 最早 offset 之和)

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 优雅重启 SOP
BROKER_ID=1
BS="10.0.0.21:9092"
KAFKA_HOME=/opt/kafka

echo "=== 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 选举(可选,减少停机抖动)==="
# 若该 Broker 只有少部分 Leader,可直接重启
# 若 Leader 很多,建议先迁移

echo "=== Step 3: 停止 Broker(优雅停机)==="
systemctl stop kafka
# 或 kill -TERM <pid>

echo "=== Step 4: 等待进程完全退出 ==="
while pgrep -f "kafka.Kafka" > /dev/null; do sleep 1; done
echo "进程已退出"

echo "=== Step 5: 启动 Broker ==="
systemctl start kafka

echo "=== Step 6: 等待启动完成 ==="
sleep 30
tail -20 /var/log/kafka/server.log | grep -E "started|RUNNING"

echo "=== Step 7: 验证集群健康 ==="
kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions
echo "(应为空)"

重启顺序原则:

  • 逐个重启,一个完全恢复(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.gz
mv 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
# /etc/systemd/system/kafka-exporter.service
[Unit]
Description=Kafka Exporter
After=network.target kafka.service

[Service]
Type=simple
ExecStart=/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=always
RestartSec=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.jar
mkdir -p /opt/kafka-exporter
mv 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:
# ===== Broker 核心指标 =====
- 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

# ISR 收缩/扩张计数
- 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

# ===== Controller =====
- 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

# ===== JVM =====
- 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
# 通过 systemd 环境文件追加,不修改发行包脚本,也不要覆盖已有 JAAS 等参数
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
# /etc/prometheus/prometheus.yml
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
# /etc/prometheus/rules/kafka-rules.yml
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 }} 无活跃成员"

# ========== Topic 健康 ==========
- 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 }} 网络空闲率过低"

# ========== JVM ==========
- 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
# 下载 zookeeper-exporter
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

# 启动(每个 ZK 节点部署一个,通过四字命令采集)
./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
# 使用 Filebeat 采集 Kafka 日志到 ELK
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
# ===== IO 线程 =====
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.interval.ms` 不配置即可使用默认 null;不能写入字符串 null
log.flush.scheduler.interval.ms=9223372036854775807

# ===== 分区文件预分配(SSD 建议关闭)=====
log.preallocate=false

9.6 JVM 调优深入

GC 日志开启(必须!否则出问题没法查):

1
2
3
4
5
6
7
8
9
10
11
# JDK 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"

# JDK 8
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
# 统计 GC 次数与耗时
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 的常见原因:

  1. 堆太小(< 6GB 且分区数多)
  2. 元空间泄漏(大量动态类生成,如频繁创建 admin client)
  3. 第三方库调用 System.gc()(加 -XX:+ExplicitGCInvokesConcurrent 缓解)
  4. 内存泄漏(通常是应用而非 Kafka 本身)

9.7 OS 层调优回顾(关键项)

1
2
3
4
5
6
7
8
9
10
11
12
# 1. PageCache 是 Kafka 性能的根本
free -g # 确保 buff/cache 很大

# 2. 磁盘调度器(SSD 用 none/noop,HDD 用 mq-deadline)
echo none > /sys/block/nvme0n1/queue/scheduler
# 永久生效需写入 udev 规则

# 3. 预读(Kafka 顺序读,适当预读有利)
blockdev --setra 4096 /dev/nvme0n1 # 预读 2MB

# 4. 网络中断亲和性(多队列网卡绑核)
# 略,需按具体网卡型号处理

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
# 1. Broker 列表与状态
/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep '(id:'

# 2. 副本不足分区(最重要的健康指标)
/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka1:9092 \
--describe --under-replicated-partitions

# 3. 消费者组消费滞后
/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 \
--all-groups --describe > /tmp/cg-$(date +%F-%H%M).txt

# 4. Broker 线程/内存快照(对每个 Broker 执行)
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

# 5. 系统层
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

# 6. 日志归档(便于事后分析)
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
# 启动失败先看这个文件,90% 的答案在这里
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 /data
echo "== 权限 =="; ls -ld /data/kafka; id kafka
echo "== 残留进程 =="; ps -ef | grep -v grep | grep kafka
echo "== ZK 连通 =="; for h in zk1 zk2 zk3; do nc -zv $h 2181; done
echo "== 内存 =="; free -g
echo "== 日志尾部 =="; 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
# ① 是否被 OOM Killer 杀死(最常见!)
dmesg -T | grep -i "killed process" | tail -20
grep -i "oom" /var/log/messages | tail -20

# ② 是否被 systemd 重启(配置 Restart=always 但进程退出)
systemctl status kafka
journalctl -u kafka --since "1 hour ago" | grep -i -E "exit|fail|kill"

# ③ 是否 GC 停顿过长导致 ZK 会话超时(ZK 认为 Broker 掉线,实际进程还活着)
grep -E "Session expired|zookeeper session" /opt/kafka/logs/server.log | tail
# GC 日志里找最长停顿
grep -E "Full GC|Pause" /opt/kafka/logs/gc.log | awk '{print $NF}' | sort -rn | head

# ④ 是否磁盘 I/O hang 导致心跳超时
iostat -x 1 3

典型的”假死”场景:Broker 进程还在,但 ZK 会话超时被踢出,导致其上的分区全部触发 Leader 切换。原因是 zookeeper.session.timeout.ms(默认 18000)小于 GC 最长停顿或磁盘 hang 时间。

处置:

  1. 加大 zookeeper.session.timeout.ms 到 18000~30000;
  2. 优化 GC 参数(见 9.6),把 Full GC 压到 1s 以内;
  3. 若磁盘存在 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
# Step 1: 确认 Broker 是否健康
/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka1:9092 \
--describe --under-replicated-partitions # 有输出说明副本有问题

# Step 2: 确认是否 Broker 侧资源饱和
# JMX: kafka.network:type=RequestChannel,name=RequestQueueSize
# 持续 > 100 说明网络/IO 线程跟不上

# Step 3: 确认客户端参数
# acks=all + min.insync.replicas=2 但 ISR 只有 1 → 必然超时

常见组合原因与处置:

组合 现象 处置
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"
# 若停在 RecordAccumulator.append → 说明 buffer.memory 满了(Broker 太慢)
# 若停在 NetworkClient.poll → 说明网络/Broker 无响应

最常见原因:某个分区 Leader 不可用,消息被粘在 RecordAccumulator 里按分区排队,导致整个批次都发不出去。
验证:查该 Topic 是否有分区 Leader = -1(--describe 输出中 Leader: none)。

10.4.3 RecordTooLargeException

三方参数必须同时满足,缺一不可:

1
2
3
4
5
6
7
8
9
10
# Broker(server.properties)
message.max.bytes=10485760 # Record Batch 上限;Kafka 3.6.2 默认 1048588

# Topic 级(覆盖 Broker 默认)
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 必然失败。

处置:

  1. 调大 max.poll.interval.ms;
  2. 减小 max.poll.records;
  3. 消费逻辑异步化,保证 poll 循环不停顿。

10.4.6 消费积压(LAG 持续增长)

判定公式:

1
LAG = (Topic 最新 offset 之和) - (消费组已提交 offset 之和)

排查决策树:

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
# 1. 加消费者实例(最直接)
# 注意:超过分区数的实例不参与消费!

# 2. 加分区(仅对新数据的并行度有效,且会改变 key 分布)
kafka-topics.sh --bootstrap-server kafka1:9092 \
--alter --topic order-event --partitions 24

# 3. 消费者侧优化:加大 fetch 参数
fetch.min.bytes=1048576 # 攒够 1MB 再返回,降低请求数
fetch.max.wait.ms=500
max.partition.fetch.bytes=1048576
max.poll.records=500

# 4. 若是历史积压且允许跳过,重置 offset 到最新
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
# 查看分区的实际 offset 范围
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

可能原因:

  1. 消费者订阅的 Topic 拼写错误(会收到空分配);
  2. 生产者写入的分区与消费者分配的分区不重叠(如只订阅了某个分区);
  3. 消费组没有有效的已提交 offset,且 auto.offset.reset=latest,因此从分区末尾开始;已有有效 offset 时该参数不生效;
  4. 消息被事务标记为未提交(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
# 找出落后的分区和落后的 broker
kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions

# 输出示例
# Topic: order-event Partition: 3 Leader: 2 Replicas: 2,3,1 Isr: 2,1
# ↑ broker 3 掉出 ISR

UnderReplicatedPartitions 是关键健康指标,但为 0 不代表集群整体健康;还要同时检查离线分区、请求延迟、磁盘容量、Controller、消费积压和错误日志。该值非 0 时应暂停非必要变更。

10.5.2 UnderReplicatedPartitions 长时间不恢复

排查三步:

1
2
3
4
5
6
7
8
9
10
# 1. 掉线的 Broker 是否已恢复?(最常见:以为恢复了其实没有)
kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep -c '(id:'

# 2. 是否残留了复制限速?(极易踩坑!)
kafka-configs.sh --bootstrap-server kafka1:9092 --describe --broker 1
# 输出中若有 leader.replication.throttled.rate / follower.replication.throttled.rate
# 且值很小(如 1MB/s),副本会同步到天荒地老

# 3. 是否 Topic 级限速?
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
# 还要清 Topic 级的
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
# server.properties(动态修改即可,无需重启)
unclean.leader.election.enable=true
1
2
3
4
5
6
# 动态修改集群默认值(对未设置 Topic 覆盖项的分区生效)
kafka-configs.sh --bootstrap-server kafka1:9092 --alter --entity-type brokers \
--entity-default --add-config unclean.leader.election.enable=true
# 或仅对单个 Topic
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 持续增长。

可能原因:

  1. Controller 本身不稳定(频繁重选举);
  2. 副本所在 Broker 的 ZK 会话超时(假死);
  3. Leader 副本所在磁盘 I/O hang,导致其被 Controller 判定为不可用。

排查:

1
2
3
4
5
6
# 查看 Controller 编号与是否频繁变化(epoch 每次切换 +1)
/opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controller
# 输出:{"version":1,"brokerid":2,"timestamp":"1694000000000"} ← brokerid 频繁变化即异常

# controller.log 里看选举原因
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
# 对比生产端与消费端的消息数(需业务侧埋点配合)
# 或对关键消息做端到端核对:生产时埋入唯一 ID,消费后落地,再比对
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
# Broker
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
# 当前 Controller
/opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka get /controller

# Controller 日志(只在 Controller 所在 Broker 上有内容)
tail -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
# perf-diagnose.sh —— 三分钟定位 Kafka 性能瓶颈
echo "===== 1. 磁盘 I/O ====="
iostat -x 1 3 | awk '$1 !~ /^(Linux|Device|loop|$) {print}'
echo "===== 2. CPU ====="
top -bn1 | head -12
echo "===== 3. 网络 ====="
sar -n DEV 1 3 | grep -E "Average|IFACE"
echo "===== 4. 内存/swap ====="
free -m
echo "===== 5. JVM GC ====="
jstat -gcutil $(pgrep -f kafka.Kafka) 1000 3
echo "===== 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 -oPa
# 找出哪些分区文件在写
ls -lhS /data/kafka/order-event-*/ | head -20

# 若是"清理线程"在抢 IO
# 检查 log.retention.check.interval.ms(默认 300000,5 分钟一次)
# 一次性清理大量段会导致 IO 尖峰,可适当调大间隔、并限制清理批量

10.7.4 句柄耗尽

1
2
3
# Kafka 打开的文件数 ≈ 分区数 × 副本数 × 段数 × 2(.log + .index)
ls /proc/$(pgrep -f kafka.Kafka)/fd | wc -l
# 报错:Too many open files

解决:调大 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

# ② 按 Topic 聚合(推荐)
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. 缩短保留期(安全,立即生效需等检查周期):
    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 # 1 天
    # 想立即触发清理,把检查周期临时调小
    kafka-configs.sh --bootstrap-server kafka1:9092 --alter --entity-type brokers \
    --entity-name 1 --add-config log.retention.check.interval.ms=60000
  2. 限制单分区大小:retention.bytes=107374182400
  3. 迁移分区到其他磁盘/节点(见第 12 章)
  4. 临时扩容磁盘(云盘在线扩容 + xfs_growfs)
  5. 最后手段:手动删除最旧 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
# 1. 检查磁盘健康
smartctl -a /dev/sdb | grep -E "Reallocated|Pending|Health"
dmesg -T | grep -i -E "I/O error|medium error"

# 2. 排查该盘上的分区
kafka-log-dirs.sh --bootstrap-server kafka1:9092 --broker-list 1 \
--topic-list order-event --describe | python3 -m json.tool

# 3. 处置:把坏盘上的分区迁走 → 下线该盘 → 换盘
# 先迁移(见 6.4),再修改 log.dirs 去掉坏盘路径,重启 Broker

警告:绝不能在 Broker 运行中直接拔盘/umount。必须先把该盘上的分区迁走,否则对应分区会进入 Offline。

10.8.3 数据目录被误删

能恢复吗? 大概率不能。Kafka 的数据是”日志”,没有 Kafka 自身的备份机制。

能做的:

  1. 如果还有其他副本存活 → 它们会自动补齐(前提是 Leader 没被删);
  2. 若所有副本都被删 → 该分区数据永久丢失,只能从上游数据源重放;
  3. 所以:数据目录的权限、备份策略、防误删(chattr +i 保护父目录)非常重要。
1
2
3
# 保护数据目录父级,防止误删(可选加固手段)
chattr +i /data # 解除:chattr -i /data
# 注意:会影响 Kafka 创建新目录,需评估后再用

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 # 查看 Kafka 进程与 JVM 参数
jstack <pid> | grep -A 30 "RequestHandler" # 请求处理线程
jstack <pid> | grep -c "java.lang.Thread.State" # 线程总数
cat /proc/<pid>/status | grep -E "Threads|VmRSS"

# ========== GC ==========
jstat -gcutil <pid> 1000 10 # GC 概览
jstat -gc <pid> 1000 10 # 各代详细
# 开启 GC 日志(KAFKA_JVM_PERFORMANCE_OPTS)
-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> # 单进程 IO/CPU
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' # 抓 RST

# ========== Kafka 自带 ==========
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

# ========== 在线诊断利器 ==========
# 1) arthas(推荐,无需重启,可看耗时/线程/反编译)
curl -O https://arthas.aliyun.com/arthas-boot.jar
java -jar arthas-boot.jar
# dashboard —— 实时面板
# thread -n 5 —— 最忙的 5 个线程
# thread -b —— 找出阻塞其他线程的元凶
# profiler start / stop --file /tmp/cpu.html —— CPU 火焰图

# 2) jmxterm(命令行读 JMX,无需 GUI)
java -jar jmxterm-1.0.4-uber.jar -l localhost:9999
# > domain kafka.server
# > bean kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
# > get Value

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 被其他进程抢走;②限速未清理是历史遗留问题

经验总结:

  1. UnderReplicatedPartitions 是第一个要看、也是最能说明问题的指标;
  2. 残留限速是极隐蔽的坑,迁移后务必确认清除;
  3. OOM Killer 是最常见的 Broker “莫名死亡”原因,dmesg 是必查项;
  4. 生产环境建议给 Kafka 进程设置 OOMScoreAdjust=-500(systemd)降低被杀的优先级。
1
2
3
4
5
6
# /etc/systemd/system/kafka.service 片段
[Service]
OOMScoreAdjust=-500
LimitNOFILE=1000000
Restart=on-failure
RestartSec=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 # 生产环境请用强密码并存入密钥管理系统

# ---------- 1. 生成 CA ----------
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"

# ---------- 2. 生成 Broker 密钥与证书(每台 Broker 都要做,key store) ----------
# 注意 CN 必须与 broker 实际 hostname 一致(客户端会校验)
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}

# 生成 CSR
keytool -certreq -alias $host -keystore ${host}.keystore.jks \
-file ${host}.csr -storepass $PASS -keypass $PASS \
-ext SAN=DNS:${host},IP:${IP}

# 用 CA 签发
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")

# 导入 CA 与签名证书到 keystore
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

# 生成 truststore(只含 CA)
keytool -importcert -alias CARoot -file ca.crt \
-keystore ${host}.truststore.jks -storepass $PASS -noprompt

done

# ---------- 3. 校验证书链和 SAN ----------
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
# ========== server.properties ==========
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
# 过渡期 Broker 间仍使用 PLAINTEXT;全部节点和凭据准备完成后再单独迁移
inter.broker.listener.name=PLAINTEXT
# 过渡期可同时保留 PLAINTEXT,切换完成后删除
# 如需内外网分流:listeners=SSL://:9093,PLAINTEXT://:9092

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

# 是否要求客户端证书(true = 双向认证,更高安全)
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
# producer.properties / consumer.properties
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
# 通过 Broker Admin API 创建用户;ZooKeeper 模式下凭据仍持久化在 ZK
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

# 查看指定用户的 SCRAM 凭据摘要(不会返回密码)
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
# ========== server.properties ==========
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

# ACL 授权器(必须显式指定!默认是 allow-all)
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
# 超级用户(不受 ACL 限制),多个用分号分隔
super.users=User:admin;User:kafka
# 是否对内部 Topic 也做鉴权(建议 true)
allow.everyone.if.no.acl.found=false

# SSL 部分同上(11.2.2)
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
# 本示例使用 SCRAM 做客户端身份认证,因此 TLS 只负责加密和服务端认证
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
# 创建 /opt/kafka/config/kafka_server_jaas.conf
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
# 加入启动参数:修改 kafka-server-start.sh 或写独立 env
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
# client-sasl.properties
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512

# 方式一:在客户端 properties 中内联 JAAS(方便,但密码仍以明文落盘)
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"

# ---------- 生产者:对 topic 写权限 ----------
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

# ---------- 前缀授权(一次授权一类 Topic,推荐!) ----------
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-

# ---------- 集群级(建 Topic 需要) ----------
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
# ZooKeeper 模式的 ACL 仍存储在 ZK;以下命令只是通过 Broker Admin API 查询,不会迁移存储位置
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
# 与现有 Broker 保持一致:JVM 版本、内核参数、Kafka 版本、目录结构
java -version
cat /etc/security/limits.conf | tail -5
sysctl -p | grep -E "swappiness|somaxconn"
df -h /data

步骤 2:拷贝安装包(保持版本完全一致)

1
2
3
# ⚠️ 必须使用与现有集群相同的 Kafka 版本,不能混版!
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
# 查看已使用的 broker.id
/opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka ls /brokers/ids
# 输出: [1, 2, 3] → 新节点可用 4

# 或通过 API
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
# 其余参数与现有一致(直接 diff 对比!)
1
2
3
# 强烈建议:直接 diff 现有配置,避免遗漏关键参数
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/kafka
chown -R kafka:kafka /data1 /data2
systemctl start kafka
tail -f /opt/kafka/logs/server.log

# 验证:新 Broker 已注册
kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | grep -c '(id:'
# 应该从 3 变成 4

此时注意:新 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

# 方式一:所有分区都重新分配(会把所有分区在所有 Broker 上重新打散,最彻底)
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

# 输出两部分:Current partition replica assignment / Proposed partition reassignment configuration
# 把 Proposed 部分保存为执行文件
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
# 先看当前分布,找出负载高的 Broker 上的分区
kafka-topics.sh --bootstrap-server kafka1:9092 --describe --topic order-event

# 输出示例(只把 broker1 上的 part-0、part-3 迁到 broker4)
# Topic: order-event Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3

cat > /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
# gen-reassign.py —— 根据各 Broker 磁盘用量生成均衡的重分配方案
python3 - <<'PY'
import json
# 目标:把分区从高负载 broker 迁到低负载 broker
# 这里给出一个简化模板,实际使用时请结合 kafka-log-dirs.sh 的输出
plan = {"version": 1, "partitions": []}
# 示例:把 broker 1 上的偶数分区迁到 broker 4
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
# 限速 50MB/s(生产建议 30~100MB/s,视带宽而定)
# ⚠️ 不限速会打满网络,导致正常的生产/消费超时甚至 ISR 收缩
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
# 反复执行,直到所有分区显示 "completed"
watch -n 30 'kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \
--reassignment-json-file /tmp/reassign-target.json --verify'

# 输出示例
# Status of partition reassignment:
# Reassignment of partition order-event-0 is completed.
# Reassignment of partition order-event-3 is still in progress.

同时盯住这两个指标(重分配期间的守护线):

1
2
3
4
5
# ① 副本不足分区数(应保持在 0,若持续 > 0 说明限速过紧或网络不足)
kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions | wc -l

# ② 磁盘写入(新 Broker 的磁盘 util 应为 60~80%,不该到 100%)
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
# 方式一:verify 显示全部 completed 后,限速会自动移除
# 方式二:手动强制清除(若中途取消或异常)
kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \
--reassignment-json-file /tmp/reassign-target.json --verify

# 手动清理残留限速(务必检查每个 Broker)
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

# 清理 Topic 级限速
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.replicas

# 确认限速已清空
for b in 1 2 3 4; do
echo "--- broker $b ---"
kafka-configs.sh --bootstrap-server kafka1:9092 --describe --broker $b | grep -i throttl
done

12.3.6 重分配完成后:Leader 均衡

重分配只改了副本位置,Leader 仍是原来的(除非原 Leader 的副本位置变了)。需要触发 Preferred Leader 选举让 Leader 回到首选位置:

1
2
3
4
5
6
7
8
9
# 对所有 Topic 执行 Preferred Leader 选举(让 Leader 分布均衡)
kafka-leader-election.sh --bootstrap-server kafka1:9092 \
--election-type PREFERRED --all-topic-partitions

# 验证 Leader 分布
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
# 理想输出:每个 broker 的 Leader 数大致相等

自动均衡更省事: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
# ===== 假设要下线 broker 4 =====

# Step 1: 确认 broker 4 上有哪些分区
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

# Step 2: 生成"把 broker4 上的副本全部搬走"的方案(写到 /tmp/remove-broker4.json)
# 官方没有直接命令,需自己生成 JSON。用下面的脚本:
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
# 用其他 broker 补位(这里简单用 1,2,3 中前 N 个;生产请按磁盘用量挑选)
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

# Step 3: 执行(限速)
kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \
--reassignment-json-file /tmp/remove-broker4.json --execute --throttle 52428800

# Step 4: 验证完成
kafka-reassign-partitions.sh --bootstrap-server kafka1:9092 \
--reassignment-json-file /tmp/remove-broker4.json --verify

# Step 5: 清除限速(对 broker 4 也要清)

# Step 6: 确认 broker 4 上已无任何分区
kafka-topics.sh --bootstrap-server kafka1:9092 --describe | grep -c "Replicas: [^ ]*4"
# 必须为 0

# Step 7: 优雅停止 broker 4
systemctl stop kafka
# 或 kill -TERM $(pgrep -f kafka.Kafka)

# Step 8: 确认 ZK 中已注销
/opt/kafka/bin/zookeeper-shell.sh zk1:2181/kafka ls /brokers/ids
# 输出不应包含 4

12.4.2 下线注意事项

  1. 绝不能直接关机!否则该节点上的所有 Leader 会被强制切换(虽然有多副本保护,但会造成短暂的元数据抖动和客户端重连)。
  2. 下线前确认 UnderReplicatedPartitions == 0。
  3. 下线过程中要监控其他节点的磁盘(接收数据会增长)。
  4. 保留下线节点的数据目录 至少 7 天,以防需要回滚。
  5. 下线后清理 ZK 中残留的配置:/config/brokers/4。

12.5 磁盘扩容与数据目录迁移

12.5.1 单机加磁盘(JBOD 模式)

1
2
3
4
5
6
7
8
9
10
11
# Step 1: 挂载新盘
mkfs.xfs /dev/sdc
mkdir -p /data3
echo "/dev/sdc /data3 xfs defaults,noatime 0 0" >> /etc/fstab
mount -a

# Step 2: 加入 log.dirs(3.x 支持多目录,Kafka 会把新分区轮询分配到新目录)
log.dirs=/data1/kafka,/data2/kafka,/data3/kafka

# Step 3: 重启 Broker(滚动重启,一次一台)
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
# replicas 顺序保持不变,只指定 broker 1 上该副本的新目录
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
# ===== 场景:/data2 磁盘即将损坏,需要换盘 =====

# Step 1: 找出该盘上的所有分区
kafka-log-dirs.sh --bootstrap-server kafka1:9092 --broker-list 1 --describe | python3 -m json.tool
# 从输出中筛选 logDir 为 /data2/kafka 的 partition 列表

# Step 2: 把这些分区的副本迁到其他盘(通过迁到其他 Broker 实现)
# Step 3: 确认迁完后,停机
systemctl stop kafka

# Step 4: 修改 log.dirs,去掉坏盘
# 第四步的含义:因为log.dirs是多路径,修改log.dirs是去掉坏盘的log.dirs路径
# Step 5: 换盘、挂载、恢复 log.dirs
# Step 6: 启动并验证
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
# /opt/kafka/config/mm2.properties
# ---------- 基础 ----------
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

# ---------- 启用 Checkpoint(关键!保证消费位点同步) ----------
source->target.enabled = true
source->target.topics = .* # 或精确列表
# 排除内部 Topic
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

# ---------- 心跳(3.x,建议单独配一个topic) ----------
source->target.emit.heartbeats.enabled = true
source->target.emit.heartbeats.interval.seconds = 5
source->target.emit.checkpoints.enabled = true
# 仅在目标消费组未处于活动状态时同步已转换的 group offset
source->target.sync.group.offsets.enabled = true

# ---------- MM2 自身配置 ----------
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
# Kafka 3.6 的专用 MM2 驱动直接读取 properties 文件
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
# ① 专用 connect-mirror-maker.sh 进程主要通过进程日志和 JMX 监控;
# 只有把 Mirror Connector 部署到独立的分布式 Kafka Connect 集群时,才使用 Connect REST API
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

# ② 监控迁移延迟(核心指标!)
# record-age-ms:源消息产生到被写入目标的时间差
# 通过 JMX: kafka.connect.mirror:type=MirrorSourceConnector,topic=.../replication-latency-ms

# ③ 常见问题
# - Consumer 报 Rebalance:加大 MM2 的 session.timeout
# - 目标集群 Topic 已存在但配置不同 → 手动对齐配置
# - 同步消费者位点不生效 → 确认 CheckpointConnector 已启动

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

# ② 健康检查(必须是 0!)
kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions | wc -l
kafka-topics.sh --bootstrap-server kafka1:9092 --describe --unavailable-partitions | wc -l

# ③ 保留当前版本二进制(用于回滚!)
cp -r /opt/kafka_2.13-3.6.2 /opt/kafka_2.13-3.6.2.bak

# ④ 确认 ZK 健康(升级过程中 ZK 必须稳定)
echo "ruok" | nc zk1 2181 # 应返回 imok
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

# ========== 阶段二:升级第一个 Broker(broker 1) ==========
# Step 1: 记录 broker1 上的 Leader 分布,必要时先迁移
kafka-topics.sh --bootstrap-server kafka1:9092 --describe | grep "Leader: 1 " | wc -l

# Step 2: 停止 broker1
systemctl stop kafka
# 确认进程退出
ps -ef | grep -c "[k]afka.Kafka" # 应为 0

# Step 3: 切换软链接到新版本(关键:先不改协议版本!)
rm -f /opt/kafka
ln -s /opt/kafka_2.13-3.7.0 /opt/kafka

# Step 4: 拷贝旧配置(新版可能新增参数项,需 merge)
cp /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.new

# Step 5: 显式固定协议版本(关键!)
cat >> /opt/kafka/config/server.properties <<'EOF'

# ===== 升级期间:使用 3.6.2 的完整 MetadataVersion,升级完成后再改 =====
inter.broker.protocol.version=3.6-IV2
EOF

# Step 6: 启动
systemctl start kafka
tail -f /opt/kafka/logs/server.log
# 看到 "started (kafka.server.KafkaServer)" 即成功

# Step 7: 验证
kafka-broker-api-versions.sh --bootstrap-server kafka1:9092 | head -3
# 这里只确认 Broker 可达与支持的 API 范围;精确二进制版本应查包管理记录、发布目录或启动日志
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
# 全部 Broker 二进制升级完成后,一次改一台:
# Step 1: 修改 broker1 的 server.properties
sed -i 's/^inter.broker.protocol.version=3.6-IV2$/inter.broker.protocol.version=3.7-IV4/' \
/opt/kafka/config/server.properties
systemctl restart kafka
# 等待 broker1 完全恢复(ISR 无异常)
sleep 300
kafka-topics.sh --bootstrap-server kafka1:9092 --describe --under-replicated-partitions | wc -l

# Step 2: broker2、broker3 重复
# Step 3: 全部完成后,检查集群
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
# 升级过程中持续执行的验证脚本
#!/bin/bash
# upgrade-watch.sh —— 升级期间每 30 秒检查一次
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 30
done

升级期间的红线:

指标 红线 动作
UnderReplicatedPartitions > 0 持续 5 分钟 暂停升级,等恢复
OfflinePartitions > 0 立即回滚
生产 P99 > 500ms 暂停升级
Broker 数 少于预期 检查刚升级的节点是否启动成功

13.4 升级回退方案

1
2
3
4
5
6
7
8
9
# 情况 1:broker 启动失败 → 直接回退二进制
systemctl stop kafka
rm -f /opt/kafka && ln -s /opt/kafka_2.13-3.6.2 /opt/kafka
cp /opt/kafka_2.13-3.6.2.bak/config/server.properties /opt/kafka/config/server.properties
systemctl start kafka

# 情况 2:已修改了 inter.broker.protocol.version 到新版本,想回退
# 提升 IBP 后可能启用旧版本不支持的元数据/协议特性,不能按普通二进制回滚处理。
# 应停止回退并按目标版本官方升级指南中的兼容边界执行恢复方案。

铁律:提升 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 小版本变化,不能使用省略参数的通用命令。

运维建议:

  1. 3.x + ZK 仅作为存量过渡架构维护,应明确迁移期限;
  2. 新集群直接使用当前受支持 Kafka 版本的 KRaft 模式;
  3. 若计划迁 KRaft,先在测试环境完整演练一次,记录每一步耗时与回退方式;
  4. 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
# /opt/kafka/bin/backup-metadata.sh
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"

# 1. Topic 定义(最关键)
$KAFKA_BIN/kafka-topics.sh --bootstrap-server $BS --describe \
> "$BACKUP_DIR/topics-describe.txt" 2>&1

# 2. Topic 配置
$KAFKA_BIN/kafka-configs.sh --bootstrap-server $BS --describe \
--entity-type topics --all > "$BACKUP_DIR/topic-configs.txt" 2>&1

# 3. Broker 动态配置
$KAFKA_BIN/kafka-configs.sh --bootstrap-server $BS --describe \
--entity-type brokers --all \
> "$BACKUP_DIR/broker-configs.txt" 2>&1

# 4. ACL
$KAFKA_BIN/kafka-acls.sh --bootstrap-server $BS --list \
> "$BACKUP_DIR/acls.txt" 2>&1 || true

# 5. 消费者组位点
$KAFKA_BIN/kafka-consumer-groups.sh --bootstrap-server $BS --all-groups --describe \
> "$BACKUP_DIR/consumer-groups.txt" 2>&1

# 6. 消费组状态(仅供审计,不能直接恢复 offset)
$KAFKA_BIN/kafka-consumer-groups.sh --bootstrap-server $BS --all-groups \
--describe --state > "$BACKUP_DIR/cg-state.txt" 2>&1

# 7. ZooKeeper 路径清单(仅供核对,不是可恢复快照)
if 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

# 8. 各 Broker 的 server.properties
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"

# 保留最近 30 天
find /data/backup/kafka-metadata -name "*.tar.gz" -mtime +30 -delete

echo "[$(date '+%F %T')] 备份完成:$BACKUP_DIR.tar.gz"

上述脚本产生的是元数据清单。另行对 ZooKeeper 的 dataDir 与 dataLogDir 做同一时间点的存储快照;若没有快照能力,应在维护窗口停止整个 ZooKeeper 集群后冷备,恢复流程必须先在测试环境演练。

1
2
# crontab:每天 3:00 执行
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
# 场景:集群元数据损坏(如 ZK 数据被误删),需要重建空集群

# 1. 部署全新的 Kafka 集群(broker.id 保持一致)
# 2. 从 Topic 汇总行中提取名称、分区数、副本数并重建
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

# 3. 按 topic-configs.txt 逐项审核并执行 kafka-configs.sh --alter
# 命令输出不是可直接执行的恢复脚本,不要用 awk/sed 自动拼接后直接上线。

# 4. 恢复 ACL
# 从 acls.txt 中逐条重新执行 kafka-acls.sh --add

# 5. 消费位点必须来自仍保留的 __consumer_offsets/MM2 checkpoint;
# 若只能重放数据,则与业务方确认后按时间或明确 offset 重置。

强烈建议:在测试环境完整演练一次”从备份重建集群”。没演练过的备份 = 没有备份。

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
# drill-broker-down.sh —— 模拟单 Broker 宕机并观察恢复
BS="kafka1:9092"
TARGET_BROKER=3

echo "===== 演练前状态 ====="
kafka-topics.sh --bootstrap-server $BS --describe --under-replicated-partitions | wc -l

echo "===== 模拟 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"
# broker 3 掉线后,其上的分区会有 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
P0(1 小时内必须恢复)
├─ 恢复 Broker 进程(让服务可用)
└─ 恢复 ZK 集群(Kafka 的元数据依赖)

P1(当天恢复)
├─ 恢复副本数到 3(消除单点)
└─ 恢复 Controller 稳定

P2(本周恢复)
├─ 补齐落后的副本
└─ 处理历史积压

P3(长期)
├─ 补齐丢失的数据(从上游重放)
└─ 复盘与预防

核心原则:先恢复可用性,再恢复完整性。不要为了”不丢一条消息”而让整个业务停摆。

14.7 数据一致性核对工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# ① 查看 Topic 的最新 offset 水位(仅用于趋势观察,不等于消息条数)
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)"

# ② 按业务主键、时间窗或消息内唯一 ID 抽样比对
# MM2 目标分区的 Kafka offset 不保证与源分区相同,不能拿相同 --offset 直接逐条对比。
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 校验日志完整性
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/kafka

# 常用命令
ls /brokers/ids # 在线 Broker 列表
get /brokers/ids/1 # 某 Broker 的详情
ls /brokers/topics # 所有 Topic
get /brokers/topics/order-event # Topic 的副本分配
get /controller # 当前 Controller
get /controller_epoch # Controller 任期
ls /admin/delete_topics # 待删除的 Topic
get /admin/reassign_partitions # 重分配进度
ls /config/topics # 有动态配置的 Topic
get /cluster/id # 集群 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 "<命令>" | nc <zk-host> <port>

# ===== 健康检查 =====
echo "ruok" | nc zk1 2181 # 返回 imok 表示存活
echo "stat" | nc zk1 2181 # 状态摘要(连接数、延迟、znode 数)
echo "srvr" | nc zk1 2181 # 服务器详情(含角色 leader/follower)
echo "conf" | nc zk1 2181 # 运行时配置

# ===== 性能指标(最重要) =====
echo "mntr" | nc zk1 2181
# 关键字段解读:
# zk_server_state leader / follower
# zk_avg_latency 平均延迟(ms),正常 < 5ms
# zk_max_latency 最大延迟(ms),关注是否 > 100ms
# zk_min_latency 最小延迟
# zk_packets_received/sent 包收发量(判断流量)
# zk_num_alive_connections 活动连接数
# zk_outstanding_requests 排队请求数(> 0 说明有积压!)
# zk_znode_count znode 总数(Kafka 大集群可达几十万)
# zk_watch_count watch 数(Kafka 大量使用 watch)
# zk_ephemerals_count 临时节点数(≈ Broker 数)
# zk_approximate_data_size 数据总量
# zk_open_file_descriptor_count 句柄数
# zk_max_file_descriptor_count 句柄上限(不够会出问题!)
# zk_followers follower 数量(仅 leader 节点有)
# zk_synced_followers 已同步的 follower 数
# zk_pending_syncs 待同步数(持续 > 0 说明有 follower 落后)

# ===== 一致性检查 =====
echo "cons" | nc zk1 2181 # 每个节点的连接/延迟详情
echo "dump" | nc zk1 2181 # 会话与临时节点列表
echo "wchs" | nc zk1 2181 # watch 统计
echo "wchc" | nc zk1 2181 # watch 详情(量大时慎用,会卡)

# ===== 调试(生产慎用!) =====
echo "envi" | nc zk1 2181
echo "srst" | nc zk1 2181 # 重置统计
# echo "kill" —— 严禁使用!
# echo "rest" —— 严禁使用!

四字命令被禁用时的处理: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,默认 18000ms)
│
├─ 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
# server.properties(Kafka 侧)
zookeeper.session.timeout.ms=18000 # 默认 18000;GC 停顿大的环境可加到 30000
zookeeper.connection.timeout.ms=18000 # 默认 null,未设置时使用 session timeout;按网络环境显式设置

# zoo.cfg(ZK 侧)
tickTime=2000
initLimit=10 # 2000 × 10 = 20s,follower 初始同步上限
syncLimit=5 # 2000 × 5 = 10s,follower 同步上限

# 约束:ZK 侧 maxSessionTimeout 必须 ≥ Kafka 的 session.timeout
# 默认 maxSessionTimeout = 20 × tickTime = 40000ms
# 所以 Kafka 侧设 30000 是安全的;若设 60000 需调大 ZK 的 maxSessionTimeout

常见坑:把 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
# 1. 确认集群状态
for h in zk1 zk2 zk3; do echo "--- $h ---"; echo "srvr" | nc -w 2 $h 2181 | grep -E "Mode|Zxid"; done

# 2. 重启宕机节点
systemctl start zookeeper

# 3. 验证其角色与数据同步
echo "srvr" | nc zk1 2181
echo "stat" | nc zk1 2181 | grep -E "Mode|Zxid|Connections"

要点:ZK 的 follower 重启后会自动从 leader 同步,不需要手动干预。但如果宕机时间过长(超过快照+事务日志保留范围),需要全量同步,会消耗较多带宽/时间。

15.5.2 ZK 失去多数派(脑裂风险)

场景:3 节点 ZK 挂了 2 个,只剩 1 个。

1
2
3
# 现象:ZK 无法选举出 leader,所有写操作失败
echo "srvr" | nc zk1 2181 | grep Mode # 显示 looking(一直找不到 leader)
# Kafka 侧表现:无法创建 Topic、Controller 选举失败、Broker 无法注册

处置(按优先级):

  1. 尽快恢复至少 1 个节点,让多数派恢复(3 节点需要 2 个);
  2. 若节点数据丢失,从备份恢复快照 + 事务日志;
  3. 绝不要通过修改 zoo.cfg 的 server 列表来”强行选主”——这会破坏 ZK 的一致性,可能导致数据分裂。

ZK 一致性公式:可用节点数 ≥ (总节点数 / 2) + 1。3 节点可挂 1 个,5 节点可挂 2 个。这就是 ZK 必须部署奇数的原因。

15.5.3 ZK 磁盘满

1
2
df -h /data/zookeeper
du -sh /data/zookeeper/version-2/* | sort -rh | head

原因:快照与事务日志未清理(autopurge 未开启或失效)。

1
2
3
# zoo.cfg
autopurge.snapRetainCount=10 # 保留 10 个快照
autopurge.purgeInterval=24 # 每 24 小时清理一次
1
2
# 使用 ZooKeeper 自带工具成组清理,保留最近 10 个快照及恢复所需事务日志
/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
# ① ZK 的 GC 是否频繁(ZK 堆不宜超过 8G)
jstat -gcutil $(pgrep -f QuorumPeerMain) 1000 5

# ② ZK 磁盘是否慢(事务日志是 fsync,对延迟极敏感)
iostat -x 1 | grep -A1 Device
# 若 dataLogDir 与 snapDir 在同一块慢盘上 → 必须分离

# ③ ZK 的 znode 数是否过多(大量 Topic、分区、ACL 和动态配置都会增加元数据)
echo "mntr" | nc zk1 2181 | grep znode_count
# 消费位点已存储在 __consumer_offsets;调整 offsets.topic.num.partitions 不能降低 ZK znode 数

# ④ watch 数是否过多
echo "mntr" | nc zk1 2181 | grep watch_count

# ⑤ 是否被 Kafka 的元数据操作打爆(如批量建 Topic)
grep -cE 'TopicChange|ConfigChange|ReassignPartitions' /opt/kafka/logs/controller.log

优化手段:

1
2
3
4
5
6
7
8
9
# zoo.cfg
tickTime=2000
# 事务日志独立磁盘(关键性能优化)
dataLogDir=/data/zk-txlog
# 关闭 fsync 前的批量(一般保持默认)
# 提高最大连接数
maxClientCnxns=200
# 关闭 AdminServer(3.5+ 默认开启,会占资源)
admin.enableServer=false
1
2
3
4
5
# JVM 参数(zkServer.sh 或 zookeeper-env.sh)
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

# 手工触发安全清理,保留最近 5 个快照及恢复所需事务日志
/opt/zookeeper/bin/zkCleanup.sh -n 5
1
2
# crontab
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
# ① 新节点安装相同版本 ZK、相同 myid 规划(4、5)
# ② 所有节点(含现有节点)修改 zoo.cfg,加入新节点
server.1=zk1:2888:3888
server.2=zk2:2888:3888
server.3=zk3:2888:3888
server.4=zk4:2888:3888
server.5=zk5:2888:3888

# ③ 滚动重启(一次一台,从 follower 开始!不要先重启 leader)
for h in zk1 zk2 zk3; do
# 先确认该节点是 follower
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
# Kafka 侧同步更新
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 # 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
# 观察 5 分钟内的 rebalance 次数
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
# 采样两次 offset,计算速率
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 60
done
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
# 升级前自动校验脚本片段:期望值来自目标版本官方升级指南,不从 API 输出猜测
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 1
fi

案例 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
# 每日巡检 OOM 检查
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
# 检查单副本 Topic(生产环境应无输出)
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
# 在 /etc/profile.d/kafka-guard.sh 中加一个函数覆盖危险命令
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
# 人工确认 plan 无误后再 --execute
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
# 通过依赖清单、服务注册信息或客户端启动日志建立版本台账。
# kafka-broker-api-versions.sh 展示的是 Broker 支持的 API 范围,不能统计客户端版本分布。

事故复盘模板(每次事故后必须填写)

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 事故预防的”十条军规”

  1. 所有 Topic 必须有明确的保留策略(retention.ms + retention.bytes)。
  2. 生产环境:副本数 ≥ 2(推荐 3),min.insync.replicas ≥ 2,acks=all,unclean.leader.election.enable=false。
  3. auto.create.topics.enable=false;Topic 删除保持可用,但必须通过最小 ACL、审批和审计受控执行。
  4. 任何迁移/重分配必须限速,且必须清除限速。
  5. 升级前必须锁定 inter.broker.protocol.version。
  6. UnderReplicatedPartitions 是最高优先级指标,长期必须为 0。
  7. Kafka 不与其他重 IO/内存型服务混部。
  8. 元数据每日备份,且每季度演练恢复。
  9. 高危命令(delete / reset-offsets / reassign / kill)纳入堡垒机审计 + 二次确认。
  10. 每次变更后 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
# Topic 参数修改示例
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
# /opt/kafka/bin/daily-check.sh —— 每日巡检
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') ==========="

# 1. Broker 数量
N=$($K/kafka-broker-api-versions.sh --bootstrap-server $BS 2>/dev/null | grep -c '(id:')
echo "Broker 在线数: ${N}"

# 2. 副本不足分区(必须 0)
UR=$($K/kafka-topics.sh --bootstrap-server $BS --describe \
--under-replicated-partitions 2>/dev/null | grep -c "Partition:")
chk "副本不足分区数为 0" "$UR"

# 3. 不可用分区(必须 0)
OP=$($K/kafka-topics.sh --bootstrap-server $BS --describe \
--unavailable-partitions 2>/dev/null | grep -c "Partition:")
chk "不可用分区数为 0" "$OP"

# 4. 副本数为 1 的 Topic(应为 0)
RF1=$($K/kafka-topics.sh --bootstrap-server $BS --describe 2>/dev/null | \
grep -oP "ReplicationFactor: \K1\b" | wc -l)
chk "单副本 Topic 数为 0" "$RF1"

# 5. 磁盘使用率(每台 < 70%)
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

# 6. 残留复制限速(应为空)
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

# 7. Controller 是否稳定(对比昨日)
CTRL=$($K/zookeeper-shell.sh zk1:2181/kafka get /controller 2>/dev/null | \
grep -o '"brokerid":[0-9]*')
echo "当前 Controller: $CTRL (与昨日对比,变化即为异常)"

# 8. ZK 状态
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

# 9. OOM Killer 记录
OOM=$(dmesg -T 2>/dev/null | grep -ci "killed process")
chk "近期无 OOM 杀进程记录" "$([ "$OOM" -eq 0 ] && echo 0 || echo 1)"

# 10. 消费者组积压(列出 LAG > 100万 的组)
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 -20

echo "=========== 巡检结束 ==========="

B.2 每周巡检

  • 磁盘增长趋势(对比上周,评估容量余量)
  • Topic 数量变化(突增说明有人乱建)
  • 消费者组数量与积压 TOP 10
  • 分区分布是否均衡(Leader 数、磁盘占用)
  • ZooKeeper 快照/事务日志磁盘占用
  • 元数据备份是否成功(检查备份文件时间戳与大小)
  • GC 日志分析(找最长停顿、Full GC 次数)
  • 慢请求 TOP 10(kafka.network:type=RequestMetrics)

B.3 每月巡检

  • 单机故障演练(kill 一台 Broker,验证自愈)
  • 副本分布检查(是否有 Broker 承担过多 Leader)
  • Topic 保留策略复核(是否存在无 retention.bytes 的 Topic)
  • 客户端版本普查(是否有过老客户端)
  • 容量预测(按增长趋势推算还能撑多久)
  • 安全配置复核(ACL 是否被误改、证书是否临近过期)
  • 参数与最佳实践比对(是否有”历史遗留”的非标准配置)

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
# /opt/kafka/bin/lag-monitor.sh —— 消费积压监控
BS="kafka1:9092"
THRESHOLD=${1:-1000000} # 告警阈值(默认 100 万)
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.txt

if [ -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/null
fi
1
2
# crontab:每 5 分钟检查一次
*/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
# /opt/kafka/bin/create-topic.sh
# 用法: create-topic.sh <topic> <partitions> <retention_days> [replication_factor]
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 1
fi
[ "$RF" -lt 2 ] && echo "❌ 生产环境副本数必须 ≥ 2" && exit 1

echo "创建 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
# /opt/kafka/bin/safe-reassign.sh
# 用法: safe-reassign.sh <reassign.json> [throttle_bytes_per_sec]
set -euo pipefail

JSON=${1:?"用法: $0 <reassign.json> [throttle]"}
THROTTLE=${2:-104857600} # 默认 100MB/s
BS="kafka1:9092"
K="/opt/kafka/bin"

# 中断时保留限速,避免仍在后台执行的重分配突然占满网络。
trap 'echo "脚本被中断;限速仍保留。恢复后重新运行本脚本或执行 --verify。" >&2; exit 130' INT TERM

echo "[$(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 60
done
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
# /opt/kafka/bin/health-check.sh
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
# /opt/kafka/bin/cleanup-logs.sh —— 清理 Kafka 自身日志文件
LOG_DIR=/opt/kafka/logs
KEEP_DAYS=7

echo "[$(date '+%F %T')] 清理 Kafka 日志(保留 $KEEP_DAYS 天)"
BEFORE=$(du -sh $LOG_DIR | cut -f1)

# 归档超过 7 天的日志
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
# /etc/systemd/system/kafka.service
[Unit]
Description=Apache Kafka Broker
Documentation=https://kafka.apache.org/documentation/
Requires=network.target remote-fs.target
After=network.target remote-fs.target zookeeper.service
Wants=zookeeper.service

[Service]
Type=simple
User=kafka
Group=kafka
Environment="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.properties
ExecStop=/bin/kill -TERM $MAINPID
Restart=on-failure
RestartSec=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
# /etc/systemd/system/zookeeper.service
[Unit]
Description=Apache ZooKeeper
Requires=network.target
After=network.target

[Service]
Type=forking
User=zookeeper
Group=zookeeper
Environment="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 start
ExecStop=/opt/zookeeper/bin/zkServer.sh stop
Restart=on-failure
RestartSec=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
# 追加到 ~/.bashrc
export KAFKA_HOME=/opt/kafka
export 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'

# 查看某 Topic 的积压
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'
}

# 查看某 Topic 在各 Broker 的分布
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
# 安装 jmxterm 后
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 官方与权威资料

资料 地址
Kafka 官方文档 https://kafka.apache.org/documentation/
Kafka 配置项全表 https://kafka.apache.org/documentation/#brokerconfigs
Kafka 协议指南 https://kafka.apache.org/protocol
KIP(改进提案)列表 https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals
Kafka 官方博客 https://kafka.apache.org/blog

D.3 生产环境检查的”终极十问”

在集群上线前,用这十个问题自查:

  1. 副本数是 3 吗?min.insync.replicas 是 2 吗?acks 是 all 吗?unclean.leader.election.enable 是 false 吗?
  2. 每个 Topic 都有明确的 retention.ms 和 retention.bytes 吗?
  3. auto.create.topics.enable 关了吗?Topic 删除是否已通过最小 ACL、审批和审计受控?
  4. ZK 是 3 或 5 节点、独立部署、事务日志在独立盘吗?
  5. Kafka 与 ZK 是否混部?Kafka 是否与其他重 IO 服务混部?
  6. UnderReplicatedPartitions 有监控和告警吗?告警能到人吗?
  7. 磁盘 70%/85% 有分级告警吗?容量能撑多久算过吗?
  8. 元数据备份开启了吗?演练过恢复吗?
  9. 高危命令(delete/reset-offsets/reassign)有审计和二次确认吗?
  10. 上一次故障演练是什么时候?演练结果有记录吗?

如果这十个问题里有任何一个答案是”没有”,那么这个集群就还没到”生产就绪”。


结语

Kafka 的运维并不复杂,难的是细节的持续执行。

这套手册里 90% 的故障,根因都可以归纳为三类:

  1. 配置不当 —— 副本数、acks、min.insync.replicas、保留策略,这四个参数决定了 80% 的稳定性;
  2. 变更失控 —— 迁移不限速、升级不锁协议版本、限速不清除、高危命令无审计;
  3. 监控缺失 —— UnderReplicatedPartitions、磁盘水位、消费积压、GC 停顿,这四个指标能提前发现绝大多数问题。

真正的运维能力,不是”出事之后能修好”,而是”让事不发生”。

最后三条建议:

  • 把所有手工操作脚本化,脚本化的操作才能被 review、被复用、被新人执行;
  • 把所有假设验证化,”我以为没问题”是事故最常见的开场白;
  • 把每次故障文档化,本手册第 16 章的每一个案例,都是某个人某天凌晨的真实经历。

版本说明:本手册基于 Apache Kafka 3.0.x ~ 3.9.x(ZooKeeper 模式)编写。
Kafka 4.0 已移除 ZooKeeper 模式。新集群使用 KRaft,存量 ZooKeeper 集群应在 3.x 阶段制定并演练迁移计划(见 13.5)。