Kafka + ZooKeeper 生产运维手册

面向生产环境的 Kafka 集群安装、配置、监控、排障、扩缩容与 ZooKeeper 协同运维指南
适用场景：日志采集、消息总线、流式计算上游、业务解耦与削峰填谷
版本说明：本文同时覆盖 ZK 协调模式（传统，Kafka ≤ 3.4）与 KRaft 模式（Kafka ≥ 3.3 推荐，3.5+ 默认方向）

---

目录

1. 消息队列基础
2. Kafka 架构与核心原理
3. 存储模型与高性能原理
4. 副本、ISR 与数据可靠性
5. Controller 与元数据管理
6. ZooKeeper 在 Kafka 中的角色
7. KRaft 模式（去 ZooKeeper）
8. 生产规划与容量评估
9. 安装部署与核心配置
10. 日常运维命令手册
11. Topic / 分区 / 副本管理
12. 消费者组与位移管理
13. 集群扩缩容与数据均衡
14. 性能调优
15. 监控告警与指标解读
16. 消息积压 / 丢失 / 重复
    • 16.1 消息积压：运维全流程
    • 16.2 消息重复消费：运维全流程
17. 安全加固（SASL / SSL / ACL）
18. 滚动升级与变更 SOP
19. ZooKeeper 运维专题
20. 故障排查手册
21. 日常巡检清单
22. 值班 SOP 速查

---

1. 消息队列基础

1.1 什么是消息队列

消息队列是一种异步的服务间通信方式，是分布式系统中重要的组件，主要解决应用耦合、异步消息、流量削峰等问题，实现高性能、高可用、可伸缩的最终一致性架构。常用产品：Kafka、RocketMQ、RabbitMQ 等。

消息中间件在消息传输过程中保存消息，充当源与目标之间的中继。队列提供路由并保证消息传递；若接收方不可用，消息会保留直到成功传递（保留期限由策略决定）。

1.2 核心组件

组件	说明
Producer	生产者，发送消息到消息队列
Consumer	消费者，从消息队列接收消息
Topic	主题，支持多订阅者；不同生产者向 Topic 发送，由 MQ 分发给订阅者
Message	消息，由描述符 + 消息体组成
Queue	队列，消息安全存放地，直到被应用处理
Broker	队列管理器，负责存储、确认、重试，包含多个队列
Channel	通道，队列管理器之间传递消息的管道

1.3 消息队列的作用

作用	说明
解耦	生产/消费独立扩展，只要遵守接口约束
削峰/缓冲	控制数据流经系统的速度，解决生产与消费速度不一致
异步通信	消息入队后不立即处理，按需消费
可扩展性	水平扩展 Producer/Consumer/Broker
持久性	消息落盘，故障后可恢复

1.4 两种消息模式

点对点（P2P）

• 消息持久化到一个队列，消费者主动拉取
• 一条消息只能被消费一次，消费后从队列删除
• 优点：拉取频率由消费者控制
• 缺点：消费者无法感知队列是否有消息，需额外监控

发布/订阅（Pub/Sub）

• 消息持久化到 Topic，消费者可订阅一个或多个 Topic
• 同一条数据可被多个消费者消费，消费完不立即删除
• 优点：消费者被动接收推送
• 缺点：无法感知各消费者处理速度差异

Kafka 属于发布/订阅模型，通过 Consumer Group 机制在同一组内实现类似 P2P 的分区独占消费。

---

2. Kafka 架构与核心原理

2.1 传统定义与特性

Kafka 是一个分布式的、基于发布/订阅模式的消息系统（事件流平台），通过高性能 TCP 协议通信，由服务器（Broker）和客户端组成。传统架构基于 ZooKeeper 协调；Kafka 3.5+ 推荐 KRaft 模式，逐步移除对 ZK 的依赖。

特性	说明
持久性、可靠性	O(1) 消息持久化；落盘 + 副本备份；消费后按保留策略删除
高吞吐、低延迟	顺序写 Page Cache；零拷贝读；批量 batch 发送
容错性	n 副本允许 n-1 节点故障；ISR 机制保障可用
高并发	支持数千客户端同时读写
可扩展性	热扩展 Broker；分区水平扩展

Kafka 优点：多生产者、多消费者（互不影响）、基于磁盘存储、伸缩性、高性能。

2.2 集群架构总览

2.3 核心概念

概念	运维视角说明
Topic	逻辑消息分类；物理上由多个 Partition 组成
Partition	有序、不可变的消息序列；Kafka 只保证分区内有序，不保证 Topic 全局有序
Replica	分区副本；分为 Leader（读写）和 Follower（同步备份）
ISR	In-Sync Replicas，与 Leader 保持同步的副本集合
Offset	分区内消息唯一递增标识；Consumer 位移即消费进度
Broker	Kafka 进程节点，无状态；数据存 log.dirs
Consumer Group	同组内一个分区只能被一个消费者消费；不同组互不影响
HW / LEO	LEO = 副本日志末端偏移；HW = 已提交可被消费者读取的最大偏移

分区分配策略（Consumer）：Range（默认）、RoundRobin、Sticky、CooperativeSticky（增量再均衡，推荐新版本）。

2.4 写消息流程

运维要点：

1. Producer 按 batch 发送，不是逐条；网络抖动时关注 batch.size、linger.ms
2. 写 Leader 所在 Broker；Follower 异步/同步拉取（由 acks 决定）
3. KRaft 模式下元数据从 Bootstrap Broker / Controller 获取，不再连 ZK

2.5 读消息流程

2.6 高性能原因

手段	原理
分布式存储	Topic 拆分为多 Partition，分散到多 Broker/磁盘
顺序写磁盘	Append-only Log，避免随机写
Page Cache	写入先落 OS 缓存，由 OS 决定刷盘时机
批量 + 压缩	Producer/Consumer/Follower 均支持 batch；lz4/snappy/zstd
零拷贝	sendfile 减少用户态/内核态拷贝
稀疏索引	.index + .timeindex + 二分查找定位 Segment

---

3. 存储模型与高性能原理

3.1 持久化结构

每个 Partition 在磁盘上是一个 append-only log 目录：

{log.dirs}/{topic}-{partition}/
├── 00000000000000000000.log      # 消息数据
├── 00000000000000000000.index    # 偏移量稀疏索引
├── 00000000000000000000.timeindex
├── 00000000000000000123.log      # 下一个 segment（默认 1GB 切分）
└── ...

• 目录命名：topic名称 + 分区序号
• 单 Segment 默认 1GB（log.segment.bytes）
• 保留策略：时间（log.retention.hours/ms）或大小（log.retention.bytes）

3.2 消息查找过程

查找 offset=221118 的消息：

1. 二分法定位所在 Segment 文件
2. 在 .index 中二分查找 ≤ 目标 offset 的最大相对 offset
3. 根据索引得到物理位置，在 .log 中顺序扫描到目标消息

运维含义：单 Partition 过大、Segment 过多会影响恢复与查找效率；磁盘随机读增多时检查是否 Page Cache 不足。

3.3 日志清理策略

策略	配置	场景
delete（默认）	cleanup.policy=delete	按时间/大小删除旧 Segment
compact	cleanup.policy=compact	按 Key 保留最新值，适合 KV/变更日志
组合	cleanup.policy=compact,delete	Compact 后再按时间清理

---

4. 副本、ISR 与数据可靠性

4.1 ACK 机制对比

方式	优点	缺点	生产建议
acks=0	低延迟、高性能	Leader 未落盘即返回，可能丢消息	仅允许丢数据的监控/埋点
acks=1	折中	Leader 宕机且 Follower 未同步则丢消息	一般业务可接受少量丢失时
acks=all/-1	最强可靠性	延迟高，依赖 ISR 副本数	金融/订单等核心链路

配合 min.insync.replicas（简称 minISR）：

• acks=all 且 ISR 副本数 < minISR → 拒绝写入（可用性换一致性）
• 生产建议：replication.factor=3，min.insync.replicas=2

4.2 ISR 机制

ISR（In-Sync Replicas）：与 Leader 保持同步的副本集合。

作用	说明
ACK 确认	acks=all 时，ISR 内全部副本确认才算成功
Leader 选举	Leader 故障时，仅从 ISR 中选举新 Leader（默认）
滞后剔除	Follower 落后超过 replica.lag.time.max.ms（默认 30s）会被踢出 ISR

延迟过高处理：Kafka 维护动态 ISR；迟迟不同步的 Follower 被剔除，避免 Leader 长时间等待。参数 replica.lag.time.max.ms 控制剔除阈值。

ISR 优缺点：

• 优点：高可靠、故障转移快、可靠性与吞吐可权衡
• 缺点：同步复制增加延迟；minISR 过低时集群不可写

4.3 关键可靠性参数

# server.properties
default.replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false   # 禁止非 ISR 副本当选 Leader，防数据丢失
replica.lag.time.max.ms=30000

# producer.properties
acks=all
enable.idempotence=true                # 幂等，防重复发送
retries=3
max.in.flight.requests.per.connection=5  # 幂等开启时 ≤5

---

5. Controller 与元数据管理

5.1 Controller 职责（ZK 模式）

集群中只有一个 Active Controller（某个 Broker 兼任），负责：

职责	说明
Leader 选举	Partition Leader 故障时触发选举
ISR 变更	维护 ISR 列表并同步到 ZK
Topic 创建/删除	分配 Partition、副本
Broker 上下线	监听 Broker 临时节点
分区重分配	协调 kafka-reassign-partitions

Controller 选举：在 ZK 上创建 /controller 临时节点，先创建成功者当选；宕机后其他 Broker 抢注。

5.2 元数据存储位置对比

模式	元数据存储	客户端获取元数据
ZK 模式	ZooKeeper + 各 Broker 缓存	Bootstrap → Metadata API
KRaft 模式	Controller Quorum 内部 Raft Log	Bootstrap → Metadata API（无 ZK）

---

6. ZooKeeper 在 Kafka 中的角色

6.1 ZK 存储的 Kafka 元数据

ZK 模式下，Kafka 在 ZK 中的典型路径（chroot 如 /kafka 时加前缀）：

ZK 路径	内容	节点类型
/brokers/ids/{id}	Broker 地址、端口、rack 等	临时
/brokers/topics/{topic}	Partition → Replica 分配	持久
/brokers/topics/{topic}/partitions/{p}/state	Leader、ISR	持久
/config/topics/{topic}	Topic 动态配置	持久
/config/changes	配置变更通知	持久
/controller	当前 Controller Broker ID	临时
/controller_epoch	Controller 纪元	持久
/admin/reassign_partitions	分区重分配状态	持久
/cluster/id	集群 ID	持久
/isr_change_notification	ISR 变更通知	持久

运维查看：

# 查看在线 Broker
zkCli.sh -server zk1:2181
ls /kafka/brokers/ids
get /kafka/brokers/ids/1

# 查看 Controller
get /kafka/controller

# 查看 Topic 分区分配
get /kafka/brokers/topics/my-topic

6.2 Broker 与 ZK 的交互

1. Broker 启动：在 /brokers/ids/{broker.id} 创建临时节点；会话超时则节点消失，触发 Controller 处理下线
2. Topic 创建：Controller 写 ZK，各 Broker 监听并创建本地 Log 目录
3. Consumer Group（旧版）：位移存 ZK；新版位移存内部 Topic __consumer_offsets（50 分区，compact）

6.3 为什么 Kafka 要移除 ZK

问题	说明
元数据双份	ZK + Broker 各维护一份，一致性与延迟复杂
分区数上限	ZK 节点数与 watch 数量限制大规模集群
运维成本	需独立维护 ZK 集群（3/5 节点）
故障域叠加	ZK 不可用影响 Controller 选举与元数据变更

→ KRaft 将元数据纳入 Kafka 自身 Raft 共识（见第 7 章）。

---

7. KRaft 模式（去 ZooKeeper）

7.1 架构变化

• Controller Quorum：通常 3 或 5 个节点（可与 Broker 同机或独立）
• 元数据写入 Raft Log，通过快照加速恢复
• Kafka 3.3+ 生产可用；3.5+ 新集群建议直接 KRaft

7.2 运维差异速查

操作	ZK 模式	KRaft 模式
元数据查询	zkCli.sh + ZK 路径	kafka-metadata.sh / kafka-metadata-quorum.sh
创建 Topic	--zookeeper（已废弃）或 --bootstrap-server	--bootstrap-server
分区重分配	支持	支持（无 ZK 参数）
配置变更	--zookeeper 或 --bootstrap-server	--bootstrap-server
监控 Controller	ZK /controller	kafka.controller:* JMX / ActiveControllerCount

7.3 KRaft 核心配置示例

# server.properties（KRaft 节点）
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@host1:9093,2@host2:9093,3@host3:9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://host1:9092
controller.listener.names=CONTROLLER
log.dirs=/data/kafka

# 格式化存储（仅首次）
kafka-storage.sh format -t <cluster-uuid> -c server.properties

# 查看 Quorum 状态
kafka-metadata-quorum.sh --bootstrap-server host1:9092 describe --status

迁移：官方提供 ZK → KRaft 迁移工具（Kafka 3.6+），生产迁移需灰度验证、备份元数据、业务低峰执行。

---

8. 生产规划与容量评估

8.1 分区数计算

目标吞吐量 Tt（MB/s）
单 Producer 吞吐 Tp，单 Consumer 吞吐 Tc

分区数 ≈ Tt / min(Tp, Tc)

示例：Tp=20MB/s，Tc=50MB/s，目标 100MB/s → 分区数 = 100/20 = 5

经验值	说明
一般 Topic	3～10 个分区起步
单分区吞吐	约 10～50 MB/s（视消息大小、副本、磁盘）
上限	过多分区增加文件句柄、选举、元数据开销；单 Broker 建议 < 4000 分区
Consumer 并行	消费实例数 ≤ 分区数 才有意义；最佳：实例数 = 分区数

8.2 Broker 数量估算

Kafka 机器数 ≈ 2 × (峰值生产速度 MB/s × 副本数 / 100) + 1

示例：峰值 50MB/s，副本 2 → 2×(50×2/100)+1 = 3 台

8.3 磁盘规划

项	建议
磁盘类型	独立 SSD/NVMe，Kafka 禁用与 OS 混用同盘
log.dirs	多目录 = 多磁盘，提升并行 IO；不同 Partition 尽量不同盘
容量	日增量 × 保留天数 × 副本数 × 1.3（余量）
文件系统	XFS / ext4；挂载 noatime
RAID	RAID10 或 JBOD（Kafka 自身有副本，常推荐 JBOD）

8.4 网络与端口

端口	用途
9092	Broker 客户端通信（可配置）
9093	KRaft Controller（KRaft 模式）
2181	ZooKeeper 客户端
2888	ZK Follower 与 Leader 通信
3888	ZK 选举

8.5 JVM 内存建议

• 堆内存：6～16GB 常见；不超过 32GB（Compressed OOPs）
• 剩余内存留给 Page Cache（Kafka 性能关键）
• 32GB 物理机：堆 6～8GB，其余给 OS Cache

---

9. 安装部署与核心配置

9.1 部署顺序（ZK 模式）

1. 部署 ZooKeeper 集群（奇数节点 3/5）
2. 部署 Kafka Broker（broker.id 唯一）
3. 验证 ZK 中 brokers/ids
4. 创建 Topic、压测、接入监控

9.2 ZooKeeper 最小配置（zoo.cfg）

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper
dataLogDir=/data/zookeeper/log
clientPort=2181
server.1=zk1:2888:3888
server.2=zk2:2888:3888
server.3=zk3:2888:3888
# 4字命令白名单（安全）
4lw.commands.whitelist=stat, ruok, conf, isro
autopurge.snapRetainCount=3
autopurge.purgeInterval=24

myid 文件：每台 ZK 的 dataDir 下写入对应数字。

9.3 Kafka server.properties 核心项

############################# Server Basics #############################
broker.id=1
process.roles=broker                    # KRaft 时: broker,controller

############################# Socket #############################
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://10.0.0.1: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 #############################
log.dirs=/data/kafka-1,/data/kafka-2
num.partitions=3
default.replication.factor=3
min.insync.replicas=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
num.recovery.threads.per.data.dir=2

############################# Replication #############################
replica.fetch.max.bytes=5242880
num.replica.fetchers=4
unclean.leader.election.enable=false

############################# Zookeeper（ZK 模式）#############################
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka
zookeeper.connection.timeout.ms=18000

############################# Topic 管理 #############################
auto.create.topics.enable=false         # 生产建议关闭，强制显式创建
delete.topic.enable=true
group.initial.rebalance.delay.ms=3000

############################# 内部 Topic #############################
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

9.4 启动与 systemd

# 前台调试
bin/kafka-server-start.sh config/server.properties

# 后台
bin/kafka-server-start.sh -daemon config/server.properties

# 验证进程
jps -l | grep Kafka
ss -lntp | grep 9092

systemd 单元示例：

[Unit]
Description=Apache Kafka
After=network.target zookeeper.service

[Service]
Type=simple
User=kafka
Environment="KAFKA_HEAP_OPTS=-Xmx8G -Xms8G"
Environment="LOG_DIR=/var/log/kafka"
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure
LimitNOFILE=100000

[Install]
WantedBy=multi-user.target

9.5 常用脚本一览

脚本	用途
kafka-topics.sh	Topic CRUD、扩分区
kafka-console-producer.sh	命令行生产
kafka-console-consumer.sh	命令行消费
kafka-consumer-groups.sh	消费组、Lag、重置位移
kafka-configs.sh	动态配置
kafka-reassign-partitions.sh	分区迁移/副本调整
kafka-preferred-replica-election.sh	Preferred Leader 选举
kafka-producer-perf-test.sh	生产压测
kafka-consumer-perf-test.sh	消费压测
kafka-log-dirs.sh	查看各 Broker 磁盘使用
kafka-dump-log.sh	解析 Log 内容

---

10. 日常运维命令手册

版本提示：Kafka 2.2+ 起管理命令逐步迁移到 --bootstrap-server；3.x 起 --zookeeper 多数已废弃。以下以 --bootstrap-server 为准，ZK 模式旧命令标注说明。

10.1 Topic 管理

# 列出 Topic（排除内部 Topic）
kafka-topics.sh --bootstrap-server broker:9092 --list --exclude-internal

# 创建 Topic：6 分区，3 副本
kafka-topics.sh --create --bootstrap-server broker:9092 \
  --topic orders --partitions 6 --replication-factor 3

# 查看详情
kafka-topics.sh --bootstrap-server broker:9092 --describe --topic orders

# 修改分区数（只增不减）
kafka-topics.sh --bootstrap-server broker:9092 \
  --alter --topic orders --partitions 12

# 删除 Topic（需 delete.topic.enable=true）
kafka-topics.sh --bootstrap-server broker:9092 --delete --topic orders

# 批量修改分区（正则）
kafka-topics.sh --bootstrap-server broker:9092 \
  --alter --topic "app-.*" --partitions 12

10.2 生产 / 消费测试

# 生产
kafka-console-producer.sh --bootstrap-server broker:9092 --topic test

# 从头消费
kafka-console-consumer.sh --bootstrap-server broker:9092 \
  --topic test --group test-g --from-beginning

# 查看消息内容并过滤
kafka-console-consumer.sh --bootstrap-server broker:9092 \
  --topic test --from-beginning | grep "keyword"

10.3 消费组与 Lag

# 列出所有消费组
kafka-consumer-groups.sh --bootstrap-server broker:9092 --list

# 查看消费详情（核心排障命令）
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group my-group

# 查看特定成员
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group my-group --members --verbose

# 删除消费组（组内无活跃成员）
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --delete --group my-group

describe 输出字段：

字段	含义
TOPIC / PARTITION	Topic 与分区号
CURRENT-OFFSET	消费者已提交位移
LOG-END-OFFSET	分区最新位移
LAG	积压 = LOG-END - CURRENT；持续增大为异常
CONSUMER-ID / HOST	消费者实例信息

10.4 查询 Topic 被哪些 Group 消费

#!/bin/bash
TOPIC="orders"
BS="broker:9092"
for g in $(kafka-consumer-groups.sh --bootstrap-server $BS --list); do
  result=$(kafka-consumer-groups.sh --bootstrap-server $BS \
    --describe --group "$g" 2>/dev/null | grep "$TOPIC")
  [ -n "$result" ] && echo "group: $g"
done

10.5 动态修改 Topic 配置

# 设置保留 1 小时
kafka-configs.sh --bootstrap-server broker:9092 \
  --alter --entity-type topics --entity-name orders \
  --add-config retention.ms=3600000

# 查看配置
kafka-configs.sh --bootstrap-server broker:9092 \
  --describe --entity-type topics --entity-name orders

# 删除动态配置（回退默认）
kafka-configs.sh --bootstrap-server broker:9092 \
  --alter --entity-type topics --entity-name orders \
  --delete-config retention.ms

# compact 策略
kafka-configs.sh --bootstrap-server broker:9092 \
  --alter --entity-type topics --entity-name changelog \
  --add-config cleanup.policy=compact

10.6 解析 Log / 查消息时间

# 查看分区日志内容
kafka-dump-log.sh --files /data/kafka/orders-0/*.log --print-data-log

# 查看各 Broker 磁盘占用
kafka-log-dirs.sh --bootstrap-server broker:9092 \
  --topic-list orders --describe

10.7 性能压测

# 生产压测：1000 万条 1KB 消息
kafka-producer-perf-test.sh \
  --topic perf-test --num-records 10000000 --record-size 1024 \
  --throughput -1 \
  --producer-props bootstrap.servers=broker:9092 acks=-1 \
  linger.ms=10 compression.type=lz4

# 消费压测
kafka-consumer-perf-test.sh \
  --bootstrap-server broker:9092 --topic perf-test \
  --messages 10000000 --threads 4

压测结论参考：

1. Producer 吞吐与 分区数正相关，与 副本数负相关
2. Consumer 吞吐与 threads 正相关，达到分区数后趋于平稳
3. 单 Broker 写入应 ≥ 50MB/s；否则检查磁盘、JVM、网络线程

---

11. Topic / 分区 / 副本管理

11.1 创建 Topic 参数说明

参数	说明
--partitions	分区数，创建后只能增加
--replication-factor	副本数，≤ Broker 数
--config	如 retention.ms=86400000,cleanup.policy=delete
--if-not-exists	幂等创建

11.2 分区扩容注意

• 只增加不减少；新分区无历史消息
• 无 Key 时新消息轮询到新分区；有 Key 时仅新 Key 可能进新分区
• 扩容后若需数据重平衡，需业务侧重新投递或自定义迁移

11.3 副本扩容（增加 replication-factor）

步骤：

# 1. 编写 reassignment JSON
cat > expand-replicas.json <<'EOF'
{
  "version": 1,
  "partitions": [
    {"topic": "orders", "partition": 0, "replicas": [1, 2, 3]},
    {"topic": "orders", "partition": 1, "replicas": [2, 3, 1]}
  ]
}
EOF

# 2. 执行（ZK 模式旧写法加 --zookeeper）
kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --reassignment-json-file expand-replicas.json --execute

# 3. 验证
kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --reassignment-json-file expand-replicas.json --verify

Leader 分布：JSON 中 replicas 第一个为 Preferred Leader；应打散到各 Broker。

11.4 Broker 扩容

1. 新机器安装 Kafka，broker.id 唯一，zookeeper.connect 指向现有集群
2. 启动 Broker，确认 ZK /brokers/ids 出现新 ID
3. 数据不会自动均衡 → 使用 kafka-reassign-partitions.sh 迁移分区
4. 可选：kafka-leader-election.sh 或 preferred-replica-election 平衡 Leader

生成迁移计划：

cat > topics-to-move.json <<'EOF'
{"topics": [{"topic": "orders"}, {"topic": "logs"}], "version": 1}
EOF

kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --topics-to-move-json-file topics-to-move.json \
  --broker-list "1,2,3,4,5" --generate

11.5 Preferred Leader 均衡

Leader 不均衡会导致热点 Broker：

# 全集群触发 Preferred Leader 选举
kafka-leader-election.sh --bootstrap-server broker:9092 --election-type preferred

# 指定分区
kafka-leader-election.sh --bootstrap-server broker:9092 \
  --election-type preferred --topic orders --partition 0

---

12. 消费者组与位移管理

12.1 再均衡（Rebalance）

触发条件：Consumer 加入/退出、订阅 Topic 变化、分区数变化。

策略	说明
Range	按 Topic 范围分配，可能不均匀
RoundRobin	轮询，要求订阅相同 Topic 集合
Sticky	尽量保持原分配，减少位移丢失
CooperativeSticky	增量再均衡，减少 Stop-The-World

生产建议：partition.assignment.strategy=CooperativeStickyAssignor（需客户端支持）。

12.2 重置位移

前提：消费组处于 Empty 或 Dead 状态（无活跃 Consumer）。

# 回到最早
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-group --reset-offsets --all-topics --to-earliest --execute

# 跳到最新（跳过积压）
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-group --reset-offsets --all-topics --to-latest --execute

# 指定 offset
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-group --reset-offsets --topic orders:0 --to-offset 10000 --execute

# 按时间（注意时区，国内常需减 8 小时或指定 TZ）
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-group --reset-offsets --topic orders \
  --to-datetime 2025-06-20T12:00:00.000 --execute

# 相对偏移
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-group --reset-offsets --topic orders --shift-by -100 --execute

# 仅导出方案不执行
kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group my-group --reset-offsets --all-topics --to-earliest --export

12.3 位移存储

• 现代 Kafka：位移写在内部 Topic __consumer_offsets（compact）
• 查看：kafka-console-consumer.sh --bootstrap-server broker:9092 \ --topic __consumer_offsets --formatter kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter

---

13. 集群扩缩容与数据均衡

13.1 缩容 Broker（谨慎）

1. 将该 Broker 上所有 Partition 迁移走（reassign）
2. 确认无 Leader / 无副本后再停进程
3. ZK 临时节点自动删除

13.2 限流迁移

大集群迁移时限制带宽，避免打满网络：

kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --reassignment-json-file plan.json --execute \
  --throttle 50000000   # 50 MB/s per broker

# 迁移完成后务必取消限流
kafka-reassign-partitions.sh --bootstrap-server broker:9092 \
  --reassignment-json-file plan.json --additional --execute --throttle -1

13.3 Cruise Control（可选）

LinkedIn 开源的 Kafka 集群自动均衡工具，支持：

• 分区/Leader 自动均衡
• Broker 故障感知
• 目标导向的容量规划

适合 50+ Broker 的大规模集群；中小集群用手动 reassign 即可。

---

14. 性能调优

14.1 磁盘与目录

• 不同 Partition 分布在不同磁盘；同盘多 Partition 会导致 IO 调度碎片化，破坏顺序写优势

14.2 JVM 参数（G1）

# bin/kafka-server-start.sh 或环境变量 KAFKA_HEAP_OPTS
KAFKA_HEAP_OPTS="-Xmx16G -Xms16G \
  -XX:+UseG1GC \
  -XX:MaxGCPauseMillis=50 \
  -XX:G1HeapRegionSize=16M \
  -XX:MetaspaceSize=256M"

G1 适用：堆 ≥ 4GB、频繁分配释放、对 GC 停顿敏感。

原则：堆不要过大，留给 Page Cache；Kafka 瓶颈多在磁盘与网络，不在 JVM 计算。

14.3 刷盘策略

Kafka 依赖 OS Page Cache，不建议强依赖 log.flush.* 强制刷盘（牺牲性能换持久性时另议）：

参数	说明
log.flush.interval.messages	累积 N 条刷盘
log.flush.interval.ms	定时刷盘
log.flush.scheduler.interval.ms	刷盘检测间隔

断电风险靠 副本 + acks=all 解决，而非单 Broker 强制刷盘。

14.4 Broker 线程与网络

num.network.threads=8          # 建议 CPU 核数 + 1
num.io.threads=16                # 建议 CPU 核数 × 2～3
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
replica.socket.receive.buffer.bytes=1048576

14.5 副本拉取

num.replica.fetchers=4
replica.fetch.min.bytes=1
replica.fetch.max.bytes=5242880    # 5MB
replica.fetch.wait.max.ms=500

14.6 Producer / Consumer 调优摘要

Producer：

acks=all
compression.type=lz4
batch.size=32768
linger.ms=10
buffer.memory=67108864
max.in.flight.requests.per.connection=5
enable.idempotence=true

Consumer：

fetch.min.bytes=1048576
fetch.max.wait.ms=500
max.poll.records=500
max.partition.fetch.bytes=1048576
enable.auto.commit=false          # 业务处理完再提交

---

15. 监控告警与指标解读

15.1 核心 JMX / Prometheus 指标

指标	含义	告警建议
UnderReplicatedPartitions	ISR 副本不足的分区数	> 0 持续 5min
OfflinePartitionsCount	无 Leader 的分区	> 0 立即 P0
ActiveControllerCount	活跃 Controller 数	≠ 1 立即 P0
RequestHandlerAvgIdlePercent	请求处理线程空闲率	< 0.2 扩容/调优
BytesInPerSec / BytesOutPerSec	入站/出站流量	突增/突降
MessagesInPerSec	消息入站速率	与基线对比
FailedFetchRequestsPerSec	Fetch 失败	持续 > 0
FailedProduceRequestsPerSec	Produce 失败	持续 > 0
IsrShrinksPerSec / IsrExpandsPerSec	ISR 收缩/扩张	频繁收缩查副本滞后
LogFlushRateAndTimeMs	刷盘耗时	磁盘异常
Consumer Lag	各分区 LAG	超过阈值或持续增长

15.2 监控架构

常用组件：

• jmx_exporter + Kafka JMX 模板
• kafka_exporter（Topic/Consumer Lag）
• Burrow（LinkedIn 消费延迟监控）

15.3 Grafana 看板关注

1. 集群总吞吐量、各 Broker 是否均衡
2. Under-Replicated / Offline Partition
3. Request 队列耗时（TotalTimeMs Produce/Fetch）
4. GC 停顿与堆使用
5. 磁盘使用率、log.dirs 各目录是否均衡
6. Top N Lag 消费组

15.4 日志关键字

关键字	可能原因
ERROR Error while writing to checkpoint	磁盘满或权限
WARN [ReplicaFetcherThread]	副本同步滞后
ERROR Fatal error during KafkaServer startup	配置错误、ZK 不可达、端口占用
WARN Connection to node -1 could not be established	网络、advertised.listeners 错误
ERROR Failed to create checkpoint	磁盘 IO 异常

---

16. 消息积压 / 丢失 / 重复

本章聚焦 运维可独立执行 的发现、定界、处置与验证；涉及业务逻辑、代码改造的部分明确标注需研发协同。

---

16.1 消息积压：运维全流程

16.1.1 什么叫积压、什么时候算故障

概念	说明
LAG	LOG-END-OFFSET − CURRENT-OFFSET，该分区未被消费的消息条数
总积压	各分区 LAG 之和（注意：条数≠字节数，大消息 Topic 需看流量）
正常	LAG 有波动但能回落；低峰时趋近 0
异常	LAG 持续单调上升 > 15～30min，或超过业务约定阈值
紧急	LAG 增速导致 磁盘/保留期内将写满，或下游 SLA 已超时

告警阈值参考（按业务调整）：

级别	条件	动作
P3 提示	单分区 LAG > 1 万 或 总 LAG > 10 万	记录、观察 15min
P2 告警	LAG 15min 增长率 > 0 且总量翻倍	启动排查 SOP
P1 紧急	LAG 影响核心交易/磁盘 > 80%	值班 + 研发 + 限流/扩容

16.1.2 第一步：5 分钟内拿到现场数据

BS="10.0.0.1:9092,10.0.0.2:9092,10.0.0.3:9092"
GROUP="order-consumer"
TOPIC="orders"

# 1. 消费组整体状态（STATE 应为 Stable，若为 PreparingRebalance 说明在抖动）
kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group $GROUP --state

# 2. 各分区 LAG 明细（核心）
kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group $GROUP | tee /tmp/lag-$(date +%Y%m%d%H%M).txt

# 3. 消费者实例是否齐全（members）
kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group $GROUP --members --verbose

# 4. Topic 分区数 vs 消费者数
PARTITIONS=$(kafka-topics.sh --bootstrap-server $BS --describe --topic $TOPIC \
  | grep -c "Partition:")
MEMBERS=$(kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group $GROUP --members 2>/dev/null | grep -c "consumer-")
echo "分区数=$PARTITIONS 在线消费者=$MEMBERS"

# 5. 生产速率（Broker JMX 或 Prometheus BytesInPerSec）
# 6. 消费服务资源：CPU/内存/GC/下游 DB 连接池（交给对应服务 owner）

输出解读速查：

现象	含义	优先怀疑
所有分区 LAG 均匀上升	消费能力整体不足	实例数、处理慢、fetch 太小
个别分区 LAG 特别高	热点分区 / Key 倾斜	分区策略、业务热点 Key
LAG 升但 CURRENT 不动	消费者卡住或已挂	进程、GC、死锁、下游超时
LOG-END 猛涨、CURRENT 正常	生产突增	上游流量、重试风暴
频繁 PreparingRebalance	再均衡风暴	实例抖动、poll 超时
某 HOST 无 member	实例掉线	K8s 驱逐、OOM、发布

一键汇总脚本（保存为 kafka-lag-report.sh）：

#!/bin/bash
BS="${1:?用法: $0 bootstrap-server group [topic]}"
GROUP="${2:?}"
TOPIC="${3:-}"

echo "========== $(date) =========="
kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group "$GROUP" --state
echo "--- LAG 明细 ---"
kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group "$GROUP" \
  | awk 'NR==1 || /PARTITION/ {next} {lag+=$6} END {print "总LAG(约):", lag+0}'
kafka-consumer-groups.sh --bootstrap-server "$BS" --describe --group "$GROUP" \
  | sort -k6 -nr | head -20
[ -n "$TOPIC" ] && kafka-topics.sh --bootstrap-server "$BS" --describe --topic "$TOPIC" | head -5

16.1.3 第二步：定界——生产过快还是消费过慢

用两次采样估算速率（运维手算）：

生产速度 ≈ ΔLOG-END-OFFSET / Δt（各分区求和）
消费速度 ≈ ΔCURRENT-OFFSET / Δt

若 生产速度 > 消费速度 → 积压必然恶化

# 间隔 5 分钟执行两次，对比同一分区的 LOG-END 与 CURRENT
watch -n 300 "kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group $GROUP | grep orders"

定界结果	运维动作	是否需要研发
生产突增	上游限流、临时扩容 Topic、通知业务	视情况
消费者不足	水平扩容实例（≤ 分区数）	否
分区数 < 并行度需求	扩分区 + 加消费者	扩分区后 Key 有序性需研发确认
单条处理慢	调 fetch 参数、加 CPU/内存	是（优化逻辑）
Rebalance 风暴	查发布记录、调超时参数	是（改客户端配置）
Broker 慢	查 ISR、磁盘、热点 Leader	否

16.1.4 处置方案矩阵（按场景选）

场景 A：消费者实例不够（最常见、运维可自行处理）

# 前提：分区数=12，当前只有 4 个消费者 → 最多 4 路并行
# 操作：将实例扩到 12（K8s 改 replicas / 进程数）

# K8s 示例
kubectl -n prod scale deployment order-consumer --replicas=12

# 验证：每个分区都有 consumer 且 LAG 下降
kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group $GROUP --members --verbose

场景 B：分区数不足（需变更 Topic，低峰执行）

# 1. 查看当前分区
kafka-topics.sh --bootstrap-server $BS --describe --topic orders

# 2. 扩分区（只增不减）例如 6 → 12
kafka-topics.sh --bootstrap-server $BS --alter --topic orders --partitions 12

# 3. 同步扩容消费者到 12
# 4. 观察 LAG（新分区只消费增量，旧分区仍积压）
# ⚠️ 有 Key 的业务需研发确认：同一 Key 仍落在原分区

场景 C：消费慢但实例已满（运维调参 + 研发优化）

运维可先改 动态配置 或协调研发改 消费者 properties（需重启生效）：

# 增大单次拉取（注意单批处理时间不能超过 max.poll.interval.ms）
fetch.min.bytes=1048576
fetch.max.wait.ms=500
max.poll.records=500
max.partition.fetch.bytes=4194304

# 防止处理一批过久被踢出组
max.poll.interval.ms=600000
session.timeout.ms=30000

# 运维侧：看消费服务是否 OOM / Full GC
jstat -gcutil <pid> 1000 10
top -Hp <pid>

# 下游依赖：DB 慢查询、Redis 超时、HTTP 接口 RT
# → 将慢查询日志、接口监控截图给研发

场景 D：生产突增（先止血）

# 1. 确认上游：发布？促销？重试循环？（查 Producer 监控、上游 QPS）
# 2. 协调上游限流（网关 / 应用层开关）
# 3. 临时调大 Topic 保留，防止磁盘打满（不治本）
kafka-configs.sh --bootstrap-server $BS \
  --alter --entity-type topics --entity-name orders \
  --add-config retention.ms=259200000

# 4. 并行扩容消费（场景 A + B）

场景 E：百万级积压紧急抢救（需审批）

适用：磁盘告急、业务允许丢弃历史或事后补数。

方案	做法	数据风险	审批
E1 跳过积压	停消费者 → --to-latest	丢弃未消费消息	业务 + 值班长
E2 转发新 Topic	写临时消费程序只转发不处理	低，需二次消费	研发 + 运维
E3 临时旁路消费	起 N 个只写 DB 的简化消费者	中，需幂等	研发

E1 跳过积压标准操作：

# ① 停止所有该 group 的消费者（K8s scale 0 或停服务）
kubectl -n prod scale deployment order-consumer --replicas=0

# ② 确认无活跃成员
kafka-consumer-groups.sh --bootstrap-server $BS --describe --group $GROUP --state
# 应显示 EMPTY

# ③ 导出当前位移备查（必做）
kafka-consumer-groups.sh --bootstrap-server $BS --group $GROUP \
  --reset-offsets --all-topics --to-current --export > /tmp/offset-backup.csv

# ④ 跳到最新（不可逆）
kafka-consumer-groups.sh --bootstrap-server $BS --group $GROUP \
  --reset-offsets --all-topics --to-latest --execute

# ⑤ 恢复消费者，确认 LAG≈0
kubectl -n prod scale deployment order-consumer --replicas=12

E2 临时 Topic 转发（研发提供程序，运维保障资源）：

原 Topic orders (LAG 大)
    ↓ 临时 consumer（只 deserialize + forward，不做业务）
新 Topic orders-rescue (分区数加倍，如 24)
    ↓ 多实例正式 consumer 并行消费
业务处理

运维职责：创建 orders-rescue、分配足够分区、部署临时转发服务、监控新 Topic LAG。

场景 F：消费卡死（LAG 升、CURRENT 不变）

# 1. 查消费者进程是否存活
kubectl -n prod get pod -l app=order-consumer
kubectl -n prod logs --tail=200 <pod> | grep -iE "error|exception|rebalance|poll"

# 2. 常见原因：下游 DB 连接池耗尽、死锁、单条 poison message
# 3. 临时措施：重启单实例（可能触发 rebalance，低峰操作）
kubectl -n prod delete pod <pod-name>

# 4. 若某分区永久卡死：研发定位消息后，运维按分区重置位移跳过毒消息（需审批）
kafka-consumer-groups.sh --bootstrap-server $BS --group $GROUP \
  --reset-offsets --topic orders:3 --shift-by 1 --execute

16.1.5 处置后验证（必须做）

# 每 5min 执行，持续 30min
kafka-consumer-groups.sh --bootstrap-server $BS --describe --group $GROUP \
  | awk 'NR>1 {s+=$6} END {print "总LAG:", s+0}'

# 通过标准：
# - LAG 总量趋势下降或稳定
# - 各分区 LAG 无单点飙高
# - 消费组 STATE=Stable
# - 业务侧确认下游数据延迟恢复

16.1.6 积压故障记录模板

【Kafka 积压事件】
时间：
Topic / Group：
峰值 LAG（条数 / 估算 MB）：
根因：（生产突增 / 消费慢 / 实例不足 / Rebalance / 其他）
处置：（扩容 N 实例 / 扩分区 / 调参 / 跳 latest / 转发）
业务影响：
是否丢数：
后续改进：（监控阈值 / 分区规划 / 代码优化）

---

16.2 消息重复消费：运维全流程

16.2.1 重复消费怎么发现（运维侧信号）

运维通常不能直接看到「重复」，而是通过以下信号介入：

信号来源	表现	运维动作
业务告警	订单重复扣款、重复发货	立即拉 Kafka 现场
DB 告警	唯一键冲突激增	查对应时段消费组
日志	Duplicate entry、幂等拦截日志暴增	对齐时间线
监控	消费 TPS 正常但业务写入翻倍	怀疑重复消费
发布关联	发版后出现	查 Rebalance、提交方式变更

# 出现重复时，第一时间固定证据
BS="broker:9092"
GROUP="order-consumer"
TOPIC="orders"

# 消费组位移是否回退（对比监控历史截图）
kafka-consumer-groups.sh --bootstrap-server $BS --describe --group $GROUP

# 该时段是否有 Rebalance（消费者日志）
kubectl logs -l app=order-consumer --since=2h | grep -i rebalance

# 消费者实例是否重启（K8s 事件）
kubectl get events --sort-by='.lastTimestamp' | grep order-consumer

16.2.2 重复消费的两条路径（先定界）

快速定界命令：

# 1. 查 Broker 是否有多条相同内容（抽样某分区）
kafka-console-consumer.sh --bootstrap-server $BS \
  --topic orders --partition 0 --offset 1000 --max-messages 10

# 2. 若 offset 不同但内容相同 → 生产端重复
# 3. 若业务说同一 orderId 处理两次、offset 不同 → 消费端重复消费或生产重复
# 4. 查 __consumer_offsets 是否被重置过（结合变更记录）

类型	典型原因	主要责任
生产端重复	retries+无幂等、上游业务重试、网络超时重发	研发改 Producer + 运维查网络
消费端重复	自动提交后宕机、处理完未提交又消费、位移被重置、to-earliest	研发改提交逻辑 + 运维查操作记录
Rebalance 重复	处理中途 rebalance，分区移交后重新消费	研发： Cooperative + 手动提交
运维误操作	--to-earliest、误重置位移	运维流程 + 审批

16.2.3 消费端重复：运维排查清单

# □ 1. 近期是否执行过位移重置（查工单 / 命令历史）
history | grep reset-offsets

# □ 2. 消费组是否发生过 Empty → 大量重消费
kafka-consumer-groups.sh --bootstrap-server $BS --describe --group $GROUP --state

# □ 3. 消费者是否 auto-commit=true（让研发确认配置）
# 正确实践：enable.auto.commit=false，业务成功后 commitSync

# □ 4. 是否发布导致频繁重启 → Rebalance
kubectl rollout history deployment/order-consumer

# □ 5. max.poll.interval 过短 → 被踢出后重新消费未提交批次
# 日志特征：Revoke partitions / Joining group
kubectl logs <pod> | grep -iE "max poll|rebalance|revoke"

运维可做的临时止血（不替代研发根治）：

措施	说明
停止抖动源	暂停频繁发布；scale 固定实例数，避免 HPA 来回扩缩
固定分区分配	低峰重启全部实例，使 group 稳定后再观察
开启业务幂等	协调研发打开 DB 唯一键 / Redis 去重（运维保障 Redis 可用）
禁止重置位移	事件期间冻结 --reset-offsets 操作

运维不应做的：

• 在未确认原因前反复 to-earliest 重放（会放大重复）
• 只重启消费者而不看 Rebalance 日志

16.2.4 生产端重复：运维排查清单

# □ 1. Producer 是否开启幂等（让研发确认或查配置中心）
# enable.idempotence=true  且  acks=all  且  max.in.flight.requests.per.connection<=5

# □ 2. 上游是否在 Kafka 之外又做了一层重试（HTTP 超时重投）
# □ 3. 网络是否不稳定（Broker / Producer 主机间 ping、丢包、防火墙）
mtr -r broker-host 9092

# □ 4. Broker 日志是否有重复 PID/Sequence（较少见）
grep -i "duplicate" /var/log/kafka/server.log | tail -50

运维协调研发修改的标准 Producer 配置：

enable.idempotence=true
acks=all
retries=2147483647
max.in.flight.requests.per.connection=5
delivery.timeout.ms=120000

16.2.5 重复发生后的数据修复（运维协同）

步骤	执行方	内容
1. 止血	运维	稳定消费组，必要时限流生产
2. 圈定范围	研发	按 orderId / 时间窗口找出重复数据
3. 停写/降级	业务	关闭自动任务，防扩散
4. 去重	研发 + DBA	SQL 去重 / 软删除 / 冲正
5. 验证	运维	LAG 正常、业务指标恢复
6. 复盘	全员	幂等、提交方式、变更审批

DB 侧辅助查重（示例，由 DBA 执行）：

-- 按业务唯一键找重复
SELECT order_id, COUNT(*) AS cnt
FROM orders
WHERE created_at >= '2025-06-24 10:00:00'
GROUP BY order_id
HAVING cnt > 1
ORDER BY cnt DESC
LIMIT 100;

16.2.6 预防：运维侧长效措施

措施	说明
变更审批	--reset-offsets 必须工单 + 双人复核
发布规范	消费者发版选低峰；CooperativeSticky 减少 rebalance
监控	Lag 外增加「业务入库速率 vs 消费速率」对比
演练	季度演练：积压扩容、误重置回滚流程
配置基线	新 Topic 检查：分区数 ≥ 预期并行度；min.insync.replicas≥2
消费组命名	环境隔离：order-consumer-prod ≠ order-consumer-test

---

16.3 消息丢失（简要）

环节	原因	防护
Producer	acks=0/1	acks=all + minISR=2
Broker	单副本 + 磁盘损坏	replication.factor≥3
Broker	unclean.leader.election=true	设为 false
Consumer	自动提交 + 处理前宕机	手动提交，业务成功后再 commit
运维误操作	--to-latest 跳过未消费消息	审批 + 导出位移备查
OS	Page Cache 未刷盘断电	副本机制 + 多机房

---

17. 安全加固（SASL / SSL / ACL）

17.1 传输加密（SSL）

# server.properties
listeners=SSL://0.0.0.0:9093
ssl.keystore.location=/path/kafka.server.keystore.jks
ssl.keystore.password=xxx
ssl.key.password=xxx
ssl.truststore.location=/path/kafka.server.truststore.jks
ssl.truststore.password=xxx
ssl.client.auth=required

客户端配置 security.protocol=SSL 及对应 truststore。

17.2 SASL 认证

listeners=SASL_SSL://0.0.0.0:9093
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# 创建 SCRAM 用户
kafka-configs.sh --bootstrap-server broker:9092 \
  --alter --add-config 'SCRAM-SHA-512=[password=secret]' \
  --entity-type users --entity-name alice

17.3 ACL 授权

# 生产者写权限
kafka-acls.sh --bootstrap-server broker:9092 \
  --add --allow-principal User:alice --producer --topic=orders

# 消费者读 + Group
kafka-acls.sh --bootstrap-server broker:9092 \
  --add --allow-principal User:bob --consumer --topic=orders --group=order-svc

# 查看 ACL
kafka-acls.sh --bootstrap-server broker:9092 --list

鉴权环境命令需带配置：

kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --command-config config/admin.properties --describe --group my-group

17.4 安全清单

• 关闭 PLAINTEXT，使用 SASL_SSL
• auto.create.topics.enable=false
• ZK 禁止公网访问，配置 ACL（ZK 3.6+）
• 定期轮换证书与 SCRAM 密码
• 最小权限 ACL，按 Topic/Group 授权

---

18. 滚动升级与变更 SOP

18.1 版本升级顺序

1. 阅读 Release Notes 兼容性（Broker 协议、ZK 版本）
2. 升级 ZooKeeper（如需要，先 ZK 后 Kafka）
3. 逐个 Broker 滚动重启（先非 Controller，最后 Controller）
4. 验证 UnderReplicated=0、生产消费正常
5. 升级 Clients（建议与 Broker 大版本接近）

18.2 单 Broker 滚动重启

# 1. 确认无 Under-Replicated（该 Broker 上）
# 2. 可选：迁移 Leader 到其他 Broker
kafka-leader-election.sh --bootstrap-server broker:9092 --election-type preferred

# 3. 停止
bin/kafka-server-stop.sh
# 或 kill $(cat /path/kafka.pid)

# 4. 替换二进制 / 修改配置
# 5. 启动并观察日志
bin/kafka-server-start.sh -daemon config/server.properties

# 6. 等待 ISR 同步完成，再下一台

18.3 变更检查清单

变更类型	风险	操作要点
修改 log.retention	低	动态配置，观察磁盘
增加分区	中	有序 Key 业务需评估
分区重分配	高	限流、低峰、verify
重置位移	高	需业务确认，先 --export
缩容 Broker	高	先迁空数据

---

19. ZooKeeper 运维专题

19.1 ZK 角色与状态

角色	说明
Leader	事务请求决策、写操作
Follower	参与选举与投票，转发写请求
Observer	只同步不投票，扩展读（Kafka 一般不用 Observer）

状态	含义
LOOKING	选举中
LEADING	当前 Leader
FOLLOWING	跟随 Leader
OBSERVING	Observer 状态

19.2 ZK 选举流程

原则：ZXID 大者优先；ZXID 相同则 myid 大者优先。

启动选举：

1. 各节点投票给自己 (myid, zxid)
2. 接收他人投票，按规则 PK 并更新本机投票
3. 过半接受同一投票 → 当选 Leader
4. 状态切换为 LEADING / FOLLOWING

运行期 Leader 宕机：剩余节点进入 LOOKING，重新选举（同上）。

19.3 ZK 与 Kafka 故障联动

ZK 故障现象	对 Kafka 影响	处理
ZK 集群不可用	无法选 Controller、无法创建 Topic	优先恢复 ZK quorum
ZK 响应慢	Broker session 超时，假死抖动	查 ZK 磁盘、GC、网络
/controller 频繁切换	Controller 抖动，分区 Leader 震荡	查 Controller Broker 负载、ZK 连接
Broker 会话超时	Broker 临时节点消失，触发 Leader 重选	查 GC、STW、zookeeper.session.timeout.ms

19.4 ZK 日常维护

# 健康检查（需白名单 ruok）
echo ruok | nc zk1 2181   # 应返回 imok

# 状态
echo stat | nc zk1 2181

# 清理快照与日志（低峰执行）
zkCleanup.sh -n 10 /data/zookeeper

# 四字命令
echo mntr | nc zk1 2181   # 详细监控指标

autopurge 自动清理（推荐）：

autopurge.snapRetainCount=10
autopurge.purgeInterval=24

19.5 ZK 监控指标

指标	告警
zk_avg_latency	持续 > 100ms
zk_outstanding_requests	积压过多
zk_open_file_descriptor_count	接近上限
节点数 znode_count	异常增长（Kafka 分区过多时）
跟随状态	Follower 与 Leader 差距过大

19.6 ZK 部署建议

• 奇数节点 3 或 5；独立机器，不与 Kafka Broker 混部（中小规模可混部但需隔离资源）
• JVM 堆 2～4GB 足够，过大反而 GC 问题
• 磁盘独立 SSD；dataLogDir 与 dataDir 分盘更佳
• 备份 dataDir 快照（冷备）

---

20. 故障排查手册

20.1 Broker 无法启动

# 检查清单
ss -lntp | grep 9092                    # 端口占用
tail -200 logs/server.log               # 启动错误
ls -la $log.dirs                        # 磁盘权限/空间
zkCli.sh -server zk:2181 ls /           # ZK 连通
cat meta.properties                     # KRaft cluster UUID 是否一致

错误	处理
Inconsistent cluster ID	格式化错误或混用数据目录
Address already in use	改端口或杀旧进程
No readable meta.properties	KRaft 未 format 或路径错误
Timed out waiting for connection	ZK 地址/防火墙/chroot 错误

20.2 生产/消费超时

1. 检查 advertised.listeners 是否客户端可达（最常见）
2. 检查安全协议是否匹配（PLAINTEXT vs SSL）
3. 检查 Topic 是否存在、auto.create.topics.enable
4. 检查 ACL 是否拒绝

20.3 副本不同步

kafka-topics.sh --bootstrap-server broker:9092 --describe --topic xxx
# 关注 Isr 数量、Leader 分布

# Broker 日志搜 ReplicaFetcherThread
# 检查 replica.fetch.max.bytes、网络带宽、磁盘 IO

20.4 Controller 异常

# ZK 模式
zkCli.sh -server zk:2181 get /kafka/controller

# 各 Broker 日志搜 "Controller"
# ActiveControllerCount JMX 应为 1

20.5 磁盘满

1. 检查 log.retention 是否过长
2. 是否有大量小 Topic 未清理
3. 紧急：调低 retention.ms、删除废弃 Topic
4. df -h + kafka-log-dirs.sh 定位大分区

20.6 消费 Rebalance 风暴

原因	处理
max.poll.interval.ms 过小	增大至大于单批处理时间
session.timeout.ms 过小	适当增大（通常 10～45s）
Consumer 频繁 OOM 重启	调堆、减少 max.poll.records
网络抖动	检查负载均衡、连接超时

---

21. 日常巡检清单

21.1 每日

• 集群 Broker 全在线，/brokers/ids 数量正确
• ActiveControllerCount = 1
• UnderReplicatedPartitions = 0，OfflinePartitionsCount = 0
• 磁盘使用率 < 75%（log.dirs 各目录）
• Top 消费组 Lag 无持续增长
• ZK ruok → imok（ZK 模式）

21.2 每周

• Leader 分布是否均衡
• GC 日志无频繁 Full GC
• 证书有效期（SSL 环境）
• ZK 快照清理是否正常
• 慢请求 / Request 队列指标趋势

21.3 变更前

• 备份相关 Topic 配置与 reassignment 方案
• 低峰窗口 + 回滚方案
• 压测环境验证（如有）
• 通知业务方

---

22. 值班 SOP 速查

场景	第一步	关键命令
消费积压	执行 kafka-lag-report.sh 拿 LAG 总量与 TOP 分区	定界后：扩实例 → 扩分区 → 调参；紧急 --to-latest 需审批（见 16.1）
重复消费	查发布记录 + Rebalance 日志 + 是否 reset-offsets	定界生产/消费端；稳定消费组；协调研发幂等（见 16.2）
集群不可用	查 OfflinePartitions	describe --topic、Controller 日志
单 Broker 宕机	看 UnderReplicated	重启 Broker，等 ISR 恢复
ZK 不可用	ruok / stat	恢复 ZK quorum，再查 Kafka
磁盘告警	kafka-log-dirs.sh	降 retention、删废弃 Topic
生产报错	看 FailedProduceRequests	ACL、ISR、minISR
副本滞后	describe --topic Isr	查网络/磁盘、限流迁移

禁止操作（未经审批）：

• 生产环境 --to-earliest / --to-latest 全量重置位移
• unclean.leader.election.enable=true
• 直接删 log.dirs 数据目录
• 多 Broker 同时重启

---

附录 A：历史命令对照（ZK 模式旧写法）

存量集群可能仍使用以下命令，新操作请统一 --bootstrap-server。

# 旧：Topic 描述
kafka-topics.sh --describe --zookeeper zk:2181/kafka --topic test

# 旧：删除 Topic
kafka-topics.sh --delete --zookeeper zk:2181/kafka --topic test

# 旧：分区重分配
kafka-reassign-partitions.sh --zookeeper zk:2181/kafka --execute ...

# 旧：动态配置
kafka-configs.sh --zookeeper zk:2181/kafka --alter ...

# 旧：消费位移检查
kafka-consumer-offset-checker.sh --zookeeper zk:2181/kafka --group g --topic t

---

文档维护建议：随集群 Kafka 版本升级，优先验证 KRaft 相关章节；ZK 模式内容在完全迁移后可移至附录。
消费组 Lag 无持续增长

• ZK ruok → imok（ZK 模式）

21.2 每周

• Leader 分布是否均衡
• GC 日志无频繁 Full GC
• 证书有效期（SSL 环境）
• ZK 快照清理是否正常
• 慢请求 / Request 队列指标趋势

21.3 变更前

• 备份相关 Topic 配置与 reassignment 方案
• 低峰窗口 + 回滚方案
• 压测环境验证（如有）
• 通知业务方

---

22. 值班 SOP 速查

场景	第一步	关键命令
消费积压	执行 kafka-lag-report.sh 拿 LAG 总量与 TOP 分区	定界后：扩实例 → 扩分区 → 调参；紧急 --to-latest 需审批（见 16.1）
重复消费	查发布记录 + Rebalance 日志 + 是否 reset-offsets	定界生产/消费端；稳定消费组；协调研发幂等（见 16.2）
集群不可用	查 OfflinePartitions	describe --topic、Controller 日志
单 Broker 宕机	看 UnderReplicated	重启 Broker，等 ISR 恢复
ZK 不可用	ruok / stat	恢复 ZK quorum，再查 Kafka
磁盘告警	kafka-log-dirs.sh	降 retention、删废弃 Topic
生产报错	看 FailedProduceRequests	ACL、ISR、minISR
副本滞后	describe --topic Isr	查网络/磁盘、限流迁移

禁止操作（未经审批）：

• 生产环境 --to-earliest / --to-latest 全量重置位移
• unclean.leader.election.enable=true
• 直接删 log.dirs 数据目录
• 多 Broker 同时重启

---

附录 A：历史命令对照（ZK 模式旧写法）

存量集群可能仍使用以下命令，新操作请统一 --bootstrap-server。

# 旧：Topic 描述
kafka-topics.sh --describe --zookeeper zk:2181/kafka --topic test

# 旧：删除 Topic
kafka-topics.sh --delete --zookeeper zk:2181/kafka --topic test

# 旧：分区重分配
kafka-reassign-partitions.sh --zookeeper zk:2181/kafka --execute ...

# 旧：动态配置
kafka-configs.sh --zookeeper zk:2181/kafka --alter ...

# 旧：消费位移检查
kafka-consumer-offset-checker.sh --zookeeper zk:2181/kafka --group g --topic t

---