Kafka 高可用集群安装手册（ZooKeeper 模式）

版本：1.0
编写日期：2026-06-08
适用范围：Linux 服务器上的自建 Apache Kafka 高可用集群，元数据管理使用 ZooKeeper。

1. 重要说明

Apache Kafka 4.x 已移除 ZooKeeper 模式，只支持 KRaft 模式。如果明确要求使用 ZooKeeper，应使用 Kafka 3.9.x 分支。本手册以 Apache Kafka 3.9.2 为例编写。Kafka 官方文档说明，Kafka 3.5 起 ZooKeeper 已标记为 deprecated，最终 3.x 小版本继续支持 ZooKeeper 模式并提供关键修复与安全修复窗口。因此新建长期集群建议优先评估 KRaft；若因兼容性、迁移窗口或组织标准需要 ZooKeeper，可按本文部署。

参考资料：

• Apache Kafka 3.9.2 发布说明：https://kafka.apache.org/blog/2026/02/21/apache-kafka-3.9.2-release-announcement/
• Apache Kafka 下载页：https://kafka.apache.org/downloads
• Kafka 3.9 ZooKeeper 运维文档：https://kafka.apache.org/39/operations/zookeeper/
• Kafka 3.9 Java 版本说明：https://kafka.apache.org/39/operations/java-version/
• Kafka 4.0 升级说明，ZooKeeper 模式已移除：https://kafka.apache.org/40/getting-started/upgrade/

2. 集群架构

2.1 推荐拓扑

生产环境建议 ZooKeeper 与 Kafka Broker 分离部署。最小高可用规格如下：

角色	数量	主机名示例	说明
ZooKeeper	3	zk1、zk2、zk3	3 节点 ensemble，可容忍 1 台故障
Kafka Broker	3	kafka1、kafka2、kafka3	副本因子 3，可容忍 1 台 Broker 故障
客户端	按需	app1、app2	生产者、消费者、管理工具

更高 SLA 可使用 5 台 ZooKeeper，可容忍 2 台 ZooKeeper 故障。ZooKeeper 节点数量建议使用奇数，不建议使用偶数。

2.2 示例 IP 规划

请按实际环境替换。

主机名	IP	角色	机架
zk1	10.0.0.11	ZooKeeper	rack-a
zk2	10.0.0.12	ZooKeeper	rack-b
zk3	10.0.0.13	ZooKeeper	rack-c
kafka1	10.0.0.21	Kafka Broker	rack-a
kafka2	10.0.0.22	Kafka Broker	rack-b
kafka3	10.0.0.23	Kafka Broker	rack-c

2.3 端口规划

端口	组件	用途
12181	ZooKeeper	客户端连接端口
2888	ZooKeeper	follower 与 leader 通信
3888	ZooKeeper	leader 选举
19092	Kafka	Broker 客户端与内部通信端口
9999	Kafka	JMX 监控端口，可选

防火墙至少放通以上端口。生产环境应只对可信网段开放。

3. 安装前准备

以下命令以 root 用户执行，服务运行用户使用 kafka。示例同时兼容 RHEL/Rocky/CentOS 和 Ubuntu 系发行版，按系统选择对应包管理命令。

3.1 时间同步

所有节点必须保持时间同步。

RHEL/Rocky/CentOS：

yum install -y chrony
systemctl enable --now chronyd
chronyc tracking

Ubuntu：

apt-get update
apt-get install -y chrony
systemctl enable --now chrony
chronyc tracking

3.2 安装 Java

Kafka 3.9 支持 Java 8、11、17、21，但 Java 8 与 Java 11 在 Kafka 3.x 中已进入弃用阶段，生产环境建议使用 Java 17 LTS 或 Java 21 LTS。本手册使用 Java 17。

RHEL/Rocky/CentOS：

yum install -y java-17-openjdk java-17-openjdk-devel nc tar wget
java -version

Ubuntu：

apt-get update
apt-get install -y openjdk-17-jdk netcat-openbsd tar wget
java -version

3.3 创建用户与目录

所有 ZooKeeper 与 Kafka 节点执行：

groupadd kafka
useradd -g kafka -m -s /bin/bash kafka

mkdir -p /opt /etc/kafka /var/log/kafka
chown -R kafka:kafka /var/log/kafka

ZooKeeper 节点额外执行：

mkdir -p /data/zookeeper/data /data/zookeeper/log
chown -R kafka:kafka /data/zookeeper

Kafka Broker 节点额外执行：

mkdir -p /data/kafka/logs
chown -R kafka:kafka /data/kafka

3.4 主机名解析

所有节点配置 DNS 或 /etc/hosts。示例：

cat >> /etc/hosts <<'EOF'
10.0.0.11 zk1
10.0.0.12 zk2
10.0.0.13 zk3
10.0.0.21 kafka1
10.0.0.22 kafka2
10.0.0.23 kafka3
EOF

验证：

ping -c 2 zk1
ping -c 2 kafka1

3.5 系统参数

所有 Kafka Broker 节点执行：

cat >/etc/security/limits.d/kafka.conf <<'EOF'
kafka soft nofile 100000
kafka hard nofile 100000
kafka soft nproc 65535
kafka hard nproc 65535
EOF

cat >/etc/sysctl.d/99-kafka.conf <<'EOF'
vm.swappiness = 1
net.core.somaxconn = 1024
net.ipv4.tcp_max_syn_backlog = 4096
net.ipv4.tcp_keepalive_time = 300
net.ipv4.tcp_keepalive_intvl = 30
net.ipv4.tcp_keepalive_probes = 5
EOF

sysctl --system

如使用独立数据盘，建议挂载参数包含 noatime，减少额外磁盘写入。

4. 安装 Kafka 软件包

所有节点执行。为了简化依赖管理，ZooKeeper 节点也安装 Kafka 二进制包，并使用 Kafka 包内脚本启动 ZooKeeper。

cd /opt
wget https://downloads.apache.org/kafka/3.9.2/kafka_2.13-3.9.2.tgz
wget https://downloads.apache.org/kafka/3.9.2/kafka_2.13-3.9.2.tgz.sha512
sha512sum kafka_2.13-3.9.2.tgz
cat kafka_2.13-3.9.2.tgz.sha512

tar -xzf kafka_2.13-3.9.2.tgz
ln -sfn /opt/kafka_2.13-3.9.2 /opt/kafka
chown -R kafka:kafka /opt/kafka_2.13-3.9.2 /opt/kafka

确认两处 SHA512 值一致后再解压。

如果无法访问 downloads.apache.org，可从 Apache Kafka 官方下载页选择镜像站下载相同版本，并保留校验步骤。

5. 部署 ZooKeeper 集群

以下步骤只在 zk1、zk2、zk3 执行。

5.1 编写配置文件

创建 /etc/kafka/zookeeper.properties：

cat >/etc/kafka/zookeeper.properties<<EOF
tickTime=2000
initLimit=10
syncLimit=5

dataDir=/data/zookeeper/data
dataLogDir=/data/zookeeper/log
clientPort=12181
maxClientCnxns=200

autopurge.snapRetainCount=10
autopurge.purgeInterval=24

admin.enableServer=false
4lw.commands.whitelist=ruok,stat,srvr,mntr,conf

server.1=zk1:2888:3888
server.2=zk2:2888:3888
server.3=zk3:2888:3888
EOF

5.2 配置 myid

每个 ZooKeeper 节点的 myid 必须与 server.N 中的 N 对应。

zk1：

echo 1 >/data/zookeeper/data/myid
chown kafka:kafka /data/zookeeper/data/myid

zk2：

echo 2 >/data/zookeeper/data/myid
chown kafka:kafka /data/zookeeper/data/myid

zk3：

echo 3 >/data/zookeeper/data/myid
chown kafka:kafka /data/zookeeper/data/myid

5.3 创建 systemd 服务

所有 ZooKeeper 节点创建 /etc/systemd/system/zookeeper.service：

[Unit]
Description=Apache ZooKeeper
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=kafka
Group=kafka
Environment="JAVA_HOME=/usr/lib/jvm/java-17-openjdk"
Environment="KAFKA_HEAP_OPTS=-Xms1g -Xmx1g"
Environment="KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:/opt/kafka/config/log4j.properties"
ExecStart=/opt/kafka/bin/zookeeper-server-start.sh /etc/kafka/zookeeper.properties
ExecStop=/opt/kafka/bin/zookeeper-server-stop.sh
Restart=on-failure
RestartSec=10
LimitNOFILE=100000

[Install]
WantedBy=multi-user.target

如果系统的 Java 路径不同，调整 JAVA_HOME。可用 readlink -f $(which java) 查询实际路径。

5.4 启动 ZooKeeper

所有 ZooKeeper 节点执行：

systemctl daemon-reload
systemctl enable --now zookeeper
systemctl status zookeeper

5.5 验证 ZooKeeper

在任意节点执行：

# -w 表示等待时间为2s
echo ruok | nc -w 2 zk1 12181
echo ruok | nc -w 2 zk2 12181
echo ruok | nc -w 2 zk3 12181

期望返回：

imok

查看角色：

echo srvr | nc -w 2 zk1 12181 | grep Mode
echo srvr | nc -w 2 zk2 12181 | grep Mode
echo srvr | nc -w 2 zk3 12181 | grep Mode

应看到 1 个 leader 和 2 个 follower。

5.6 创建 Kafka chroot

本手册使用 ZooKeeper chroot /kafka-prod，便于隔离 Kafka 元数据。

任意 ZooKeeper 节点执行：

# 执行创建生产环境的kafka节点
echo "create /kafka-prod ''" | /opt/kafka/bin/zookeeper-shell.sh zk1:12181
# 查看zookeeper的根路径内容
echo "ls /" | /opt/kafka/bin/zookeeper-shell.sh zk1:12181

如果提示节点已存在，可忽略。

6. 部署 Kafka Broker 集群

以下步骤只在 kafka1、kafka2、kafka3 执行。

6.1 编写 Broker 配置

创建 /etc/kafka/server.properties。三台 Broker 大部分配置相同，broker.id、advertised.listeners、broker.rack 需要按节点修改。

kafka1：

cat >/etc/kafka/server.properties<<EOF
#########################################
# 1. 基础集群标识 (Broker & Rack)
#########################################
# 当前 Kafka 节点的唯一标识，在集群中必须是唯一的整数
broker.id=1
# 声明当前 Broker 所在的物理机架名称，用于机架感知副本分配以实现容灾
broker.rack=rack-a

#########################################
# 2. 网络与监听器配置 (Listeners)
#########################################
# 代理监听的网卡和端口，0.0.0.0 表示监听内网所有网卡
listeners=PLAINTEXT://0.0.0.0:19092
# 注册到 Zookeeper 并暴露给客户端（生产者/消费者）用于集群发现和连接的真实地址/域名
advertised.listeners=PLAINTEXT://kafka1:19092
# 定义监听器名称与安全协议的映射关系，这里表示 PLAINTEXT 监听器使用明文传输协议
listener.security.protocol.map=PLAINTEXT:PLAINTEXT
# 规定 Kafka 集群内部各个 Broker 之间进行数据同步和通信时使用的监听器名称
inter.broker.listener.name=PLAINTEXT

#########################################
# 3. 日志与默认副本策略 (Storage & Replication)
#########################################
# Kafka 存放消息日志（数据文件）的实际物理磁盘路径
log.dirs=/data/kafka/logs
# 创建 Topic 时如果没有显式指定分区数，默认自动创建的分区数量
num.partitions=3
# 创建 Topic 时如果没有显式指定副本因子，默认自动创建的副本数量（包含 Leader 和 Follower）
default.replication.factor=3
# 最小同步副本数。配合生产者 acks=all 使用，表示必须至少有 2 个副本确认写入成功消息才算发送成功
min.insync.replicas=2

#########################################
# 4. 内部元数据 Topic 策略
#########################################
# 存储消费者消费位移（Consumer Offsets）的内部 Topic（__consumer_offsets）的副本数
offsets.topic.replication.factor=3
# 事务状态日志 Topic 的副本数量
transaction.state.log.replication.factor=3
# 事务状态日志 Topic 的最小同步副本数（ISR）
transaction.state.log.min.isr=2

#########################################
# 5. Zookeeper 连接配置
#########################################
# Kafka 连接的 Zookeeper 集群地址，最后的 /kafka-prod 是 Chroot 路径，用于存放元数据并实现环境隔离
zookeeper.connect=zk1:12181,zk2:12181,zk3:12181/kafka-prod
# 允许连接 Zookeeper 的最大超时时间（毫秒），这里是 18 秒
zookeeper.connection.timeout.ms=18000

#########################################
# 6. 高级运营行为控制 (Safety & Operations)
#########################################
# 禁用自动创建 Topic。生产环境必设，防止客户端因拼写错误自动创建出不符合规范的 Topic
auto.create.topics.enable=false
# 允许通过管理员命令彻底删除 Topic 及其磁盘数据
delete.topic.enable=true
# 禁用不干净的 Leader 选举。当 Leader 挂掉时，不允许未完全同步数据的 Follower 竞选新 Leader，优先保证数据不丢失
unclean.leader.election.enable=false

#########################################
# 7. 线程与性能调优 (Threading)
#########################################
# 负责接收并处理网络请求的线程数量（主要负责将请求读写到内存缓冲区）
num.network.threads=3
# 实际负责将数据从内存缓冲区读写到磁盘 I/O 的线程数量（通常设为 CPU 核心数）
num.io.threads=8
# 从其他 Broker 副本节点复制/拉取数据的线程数量
num.replica.fetchers=2
# 网络线程池中允许排队等待处理的最大请求上限数量
queued.max.requests=500

#########################################
# 8. 日志保留与清理策略 (Retention)
#########################################
# 数据在磁盘上的最长保留时间，超时的数据将被清理（168 小时即 7 天）
log.retention.hours=168
# 单个日志分段（Segment）文件的最大大小，达到 1GB（1073741824 字节）后会关闭当前文件并滚动创建新文件
log.segment.bytes=1073741824
EOF

kafka2 修改：

broker.id=2
broker.rack=rack-b
advertised.listeners=PLAINTEXT://kafka2:19092

kafka3 修改：

broker.id=3
broker.rack=rack-c
advertised.listeners=PLAINTEXT://kafka3:19092

说明：

• default.replication.factor=3：默认主题副本数为 3。
• min.insync.replicas=2：配合生产者 acks=all，至少 2 个 ISR 副本确认后才认为写入成功。
• unclean.leader.election.enable=false：避免非同步副本成为 leader，降低数据丢失风险。
• auto.create.topics.enable=false：避免客户端误拼 topic 名导致自动创建错误主题。

如客户端需要通过域名、VIP、NAT 或跨网段访问，必须把 advertised.listeners 改为客户端可访问的地址。

6.2 创建 systemd 服务

所有 Kafka Broker 节点创建 /etc/systemd/system/kafka.service：

cat >/etc/systemd/system/kafka.service<<EOF
[Unit]
Description=Apache Kafka Broker
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=kafka
Group=kafka
Environment="JAVA_HOME=/usr/lib/jvm/java-17-openjdk"
Environment="KAFKA_HEAP_OPTS=-Xms2g -Xmx2g"
Environment="KAFKA_JVM_PERFORMANCE_OPTS=-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent"
Environment="JMX_PORT=9999"
ExecStart=/opt/kafka/bin/kafka-server-start.sh /etc/kafka/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure
RestartSec=10
LimitNOFILE=100000
TimeoutStopSec=120

[Install]
WantedBy=multi-user.target
EOF

如果服务器内存较小，可先将 KAFKA_HEAP_OPTS 调整为 -Xms2g -Xmx2g。生产环境建议根据分区数、连接数、吞吐量和监控结果调优。

6.3 启动 Kafka Broker

建议逐台启动，启动间隔 10 到 30 秒。

systemctl daemon-reload
systemctl enable --now kafka
systemctl status kafka

查看日志：

journalctl -u kafka -f

7. 集群验证

以下命令可在任意 Kafka Broker 节点执行。

7.1 查看 Broker API

/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092

能返回各 Broker 信息即表示基础连接正常。

7.2 创建测试 Topic

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --create \
  --topic ha-test \
  --partitions 6 \
  --replication-factor 3 \
  --config min.insync.replicas=2

查看 Topic：

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --describe \
  --topic ha-test

检查每个分区的 Replicas 有 3 个副本，Isr 正常情况下也有 3 个副本。

7.3 生产和消费测试

生产消息：

/opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --topic ha-test \
  --producer-property acks=all

输入几行消息后退出。

消费消息：

/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --topic ha-test \
  --from-beginning \
  --timeout-ms 10000

7.4 故障演练

停止一台 Broker：

systemctl stop kafka

在其他 Broker 节点查看 Topic：

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --describe \
  --topic ha-test

预期：

• 集群仍可读写。
• 部分分区 Isr 从 3 变为 2。
• 因 min.insync.replicas=2，剩余 2 个 ISR 时生产者 acks=all 仍可写入。

恢复 Broker：

systemctl start kafka

等待副本追平后，Isr 应恢复为 3。

停止一台 ZooKeeper：

systemctl stop zookeeper

3 节点 ZooKeeper 可容忍 1 台故障，Kafka 集群应继续运行。恢复：

systemctl start zookeeper

8. 生产建议

8.1 Topic 规范

生产 Topic 建议显式创建，避免依赖自动创建。

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --create \
  --topic business-events \
  --partitions 12 \
  --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000

关键建议：

• 生产 Topic 副本数使用 3。
• 重要业务生产者使用 acks=all。
• 重要业务 Topic 使用 min.insync.replicas=2。
• 分区数按吞吐量和消费者并发规划，后续只能增加，不能减少。

8.2 监控指标

Broker 重点监控：

指标	含义	建议
UnderReplicatedPartitions	副本不同步分区数	正常应为 0
OfflinePartitionsCount	离线分区数	必须为 0
ActiveControllerCount	活跃 Controller 数	全集群总和应为 1
RequestHandlerAvgIdlePercent	请求线程空闲率	长期过低需扩容或调优
BytesInPerSec、BytesOutPerSec	读写吞吐	结合磁盘和网络容量观察
NetworkProcessorAvgIdlePercent	网络线程空闲率	长期过低需扩容或增加网络线程

指标名称	指标含义	正常标准 / 告警阈值	命令行查看方法 (JMX / 脚本)	关联的 server.properties 配置
UnderReplicatedPartitions	副本不同步的分区数	必须为 0 (大于 0 说明有 Follower 节点掉线或网络卡顿)	kafka-topics.sh --bootstrap-server kafka1:19092 --describe --under-replicated-partitions	default.replication.factor=3 min.insync.replicas=2
OfflinePartitionsCount	离线分区数	必须为 0 (大于 0 说明分区已无 Leader，读写中断)	kafka-topics.sh --bootstrap-server kafka1:19092 --describe --unavailable-partitions	unclean.leader.election.enable=false
ActiveControllerCount	活跃 Controller 数量	整个集群总和必须为 1 (0 表示瘫痪，>1 表示脑裂)	kafka-run-class.sh kafka.tools.JmxTool --object-name "kafka.controller:type=KafkaController,name=ActiveControllerCount" --bootstrap-server kafka1:19092	zookeeper.connect=...
RequestHandlerAvgIdlePercent	I/O 请求处理线程空闲率	长期建议 > 0.3 (30%) (过低说明磁盘或 CPU 达到瓶颈)	kafka-run-class.sh kafka.tools.JmxTool --object-name "kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent" --bootstrap-server kafka1:19092	num.io.threads=8
NetworkProcessorAvgIdlePercent	网络处理线程空闲率	长期建议 > 0.3 (30%) (过低说明网卡吞吐或连接数占满)	kafka-run-class.sh kafka.tools.JmxTool --object-name "kafka.network:type=Processor,name=NetworkProcessorAvgIdlePercent" --bootstrap-server kafka1:19092	num.network.threads=3
BytesInPerSec	每秒输入字节数 (写入流量)	结合服务器物理网卡与磁盘写入上限观察	kafka-run-class.sh kafka.tools.JmxTool --object-name "kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec" --bootstrap-server kafka1:19092	log.segment.bytes=1073741824
BytesOutPerSec	每秒输出字节数 (读取流量)	结合服务器物理网卡带宽上限观察	kafka-run-class.sh kafka.tools.JmxTool --object-name "kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec" --bootstrap-server kafka1:19092	-

ZooKeeper 重点监控：

指标	含义	建议
ruok	健康检查	返回 imok
mntr zk_avg_latency	平均延迟	持续升高需检查磁盘和网络
mntr zk_outstanding_requests	堆积请求	持续大于 0 需关注
srvr Mode	leader/follower 状态	保持 1 leader，其余 follower

命令示例：

echo mntr | nc -w 2 zk1 12181
echo srvr | nc -w 2 zk1 12181

8.3 日志与数据目录

• ZooKeeper 的 dataLogDir 建议使用独立低延迟磁盘。
• Kafka 的 log.dirs 建议使用独立数据盘，不要与系统盘混用。
• 不要直接手工删除 Kafka 数据目录中的分区文件。
• 磁盘使用率建议在 70% 以下，超过 80% 必须扩容或降低保留时间。

8.4 滚动重启

Kafka Broker 滚动重启顺序：

1. 确认无离线分区、无欠复制分区。
2. 停止一台 Broker。
3. 等待 Controller 完成 leader 迁移。
4. 启动该 Broker。
5. 等待 UnderReplicatedPartitions 恢复为 0。
6. 继续下一台。

命令：

systemctl stop kafka
systemctl start kafka

ZooKeeper 滚动重启顺序：

1. 一次只重启一台 ZooKeeper。
2. 重启 follower 优先，leader 最后。
3. 每次重启后确认 ensemble 恢复正常。

查看角色：

echo srvr | nc -w 2 zk1 12181 | grep Mode

8.5 安全加固

本手册示例使用 PLAINTEXT，便于快速部署和验证。生产环境建议从传输加密（TLS）**、身份认证（SASL/mTLS）、**权限控制（ACL）以及网络隔离四个维度进行配置：

• Kafka 客户端通信启用 TLS：SSL 或 SASL_SSL。
• Kafka 客户端认证启用 SASL/SCRAM、Kerberos 或 mTLS。
• Kafka ACL 开启授权控制。
• ZooKeeper 开启 SASL 认证和 TLS。
• JMX 不要裸露到非可信网络，至少限制来源 IP，最好接入安全的监控代理。
• 防火墙只放通必要端口，禁止公网直接访问 Kafka 和 ZooKeeper。

具体的配置方法：

安全加固项	核心配置方法与核心参数	涉及组件 / 配置文件	补充说明与最佳实践
1. 客户端通信启用 TLS (SSL 或 SASL_SSL)	步骤： 1. 为每个 Broker 生成 KeyStore/TrustStore 证书；2. 修改监听器。 核心参数： listeners=SASL_SSL://0.0.0.0:19092 advertised.listeners=SASL_SSL://kafka1:19092 ssl.keystore.location=/path/to/server.keystore.jks ssl.keystore.password=your_password ssl.truststore.location=/path/to/server.truststore.jks ssl.truststore.password=your_password	Kafka Broker (server.properties)	* 如果只需要加密不需要客户端证书认证，用 SASL_SSL； * 内部通信建议同步改为 inter.broker.listener.name=SASL_SSL。
2. 客户端认证 (SASL/SCRAM, Kerberos, mTLS)	根据选择三选一配置： ① SASL/SCRAM (推荐，轻量)： listener.security.protocol.map=SASL_SSL:SASL_SSL sasl.enabled.mechanisms=SCRAM-SHA-512 并在 JAAS 文件中配置管理用户，使用 kafka-configs.sh 创建业务用户。 ② Kerberos (常规企业级)：配置 GSSAPI 机制并关联 krb5.conf 与 Keytab 文件。 ③ mTLS (双向证书认证)：设置 ssl.client.auth=required。	Kafka Broker (server.properties 和 kafka_server_jaas.conf)	* SCRAM-SHA-512 是目前最兼顾安全与运维便利性的认证方式； * 启动 Kafka 时需要添加 JVM 参数：-Djava.security.auth.login.config=/path/to/kafka_server_jaas.conf。
3. 开启 ACL 授权控制	核心参数： authorizer.class.name=kafka.security.authorizer.AclAuthorizer super.users=User:admin 管理命令： 开启后，使用 kafka-acls.sh 授予用户对特定 Topic 的读写权限： kafka-acls.sh --bootstrap-server kafka1:19092 --command-config admin.properties --add --allow-principal User:alice --operation Read --topic test-topic	Kafka Broker (server.properties)	* 必须先开启身份认证（如 SASL），ACL 才能根据用户名进行权限判定； * super.users 定义超级管理员，该用户不受 ACL 限制。
4. ZooKeeper 开启 SASL 认证和 TLS	① SASL 认证（防止匿名登录 zk）： authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider requireClientAuthScheme=sasl ② TLS 加密通信： secureClientPort=212181 ssl.keyStore.location=/path/to/zk.server.keystore.jks ssl.keyStore.password=password ssl.trustStore.location=/path/to/zk.server.truststore.jks ssl.trustStore.password=password	ZooKeeper (zoo.cfg 和 zookeeper_jaas.conf)	* ZooKeeper 开启 SASL 后，Kafka 的 kafka_server_jaas.conf 中也必须配置 Client 模块以连接 zk； * 开启 TLS 后，Kafka 的 server.properties 中连接地址需改为 zookeeper.connect=zk1:212181... 并配置 zookeeper.ssl.client.enable=true。
5. JMX 安全防护	做法一（限制 IP）： 配合监控代理（如 Prometheus JMX Exporter），JMX 开启 127.0.0.1 本地监听，由 Exporter 抓取指标，不暴露远程 JMX 端口。 做法二（必须远程时）： 开启 JMX 密码认证： -Dcom.sun.management.jmxremote.authenticate=true -Dcom.sun.management.jmxremote.password.file=/path/to/jmxremote.password -Dcom.sun.management.jmxremote.access.file=/path/to/jmxremote.access	Kafka 启动环境变量 (kafka-env.sh 或系统环境变量)	* 生产环境强烈不建议直接暴露未加密、无认证的 JMX_PORT 到非信任网络； * 推荐使用 Prometheus 架构，安全且易于集中管理。
6. 防火墙端口限制	Linux 防火墙策略 (以 iptables / firewalld 为例)： ① 拒绝所有外网直接访问 19092 和 12181/212181 端口； ② 仅放通特定内网堡垒机、监控服务器、应用服务器 IP。 firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="192.168.1.100" port protocol="tcp" port="19092" accept'	宿主机操作系统 (Firewall / Security Group)	* 防火墙是最后一道防线； * 如果在云环境（AWS、阿里云等），优先使用**安全组（Security Group）**将 Kafka 严格限制在 VPC 内网，绝不绑定公网 IP。

8.6 备份

建议备份内容：

• /etc/kafka/*.properties
• /etc/systemd/system/kafka.service
• /etc/systemd/system/zookeeper.service
• ZooKeeper 快照与事务日志的备份策略
• Topic、ACL、客户端配置信息导出

Kafka 消息数据本身通常通过副本和跨集群复制保障。对强合规数据，可额外设计 MirrorMaker 2、对象存储归档或下游数仓落盘。

9. 常见问题

9.1 Broker 启动失败，提示连接 ZooKeeper 失败

检查项：

echo ruok | nc -w 2 zk1 12181
echo ruok | nc -w 2 zk2 12181
echo ruok | nc -w 2 zk3 12181

确认：

• ZooKeeper 服务已启动。
• 防火墙放通 12181。
• /kafka-prod chroot 已创建。
• zookeeper.connect 中主机名可解析。

9.2 客户端连上 bootstrap server 后无法访问 Broker

通常是 advertised.listeners 配错。客户端先连 bootstrap Broker，然后会按 Broker 返回的 advertised 地址继续访问其他 Broker。因此 advertised.listeners 必须是客户端可解析且可连通的地址。

9.3 创建 Topic 时报副本数不足

如果 3 副本 Topic 创建失败，检查 3 台 Broker 是否全部在线：

/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092

9.4 生产者报 NotEnoughReplicas

通常是 ISR 数量低于 min.insync.replicas。检查：

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --describe \
  --topic ha-test

处理：

• 恢复故障 Broker。
• 检查磁盘、网络和 Broker 日志。
• 不建议为了临时恢复写入而降低 min.insync.replicas，除非业务明确接受更高的数据丢失风险。

9.5 ZooKeeper 频繁切换 leader

检查：

• ZooKeeper 节点之间网络抖动。
• ZooKeeper 事务日志磁盘延迟。
• JVM GC 停顿。
• 是否与 Kafka Broker 或其他高负载服务混部。

Kafka 官方运维建议 ZooKeeper 尽量保持简单、隔离部署，并重视事务日志磁盘性能。

10. 交付检查清单

部署完成后逐项确认：

检查项	结果
3 台 ZooKeeper 均启动	待确认
ZooKeeper 返回 1 leader、2 follower	待确认
/kafka-prod chroot 已创建	待确认
3 台 Kafka Broker 均启动	待确认
kafka-broker-api-versions.sh 可查询全部 Broker	待确认
测试 Topic 副本数为 3，ISR 为 3	待确认
停止 1 台 Broker 后集群仍可读写	待确认
恢复 Broker 后 ISR 回到 3	待确认
防火墙只放通必要端口	待确认
JMX 与日志已接入监控	待确认
关键配置文件已备份	待确认

11. 附录：一键查看状态命令

ZooKeeper：

for h in zk1 zk2 zk3; do
  echo "===== $h ====="
  echo ruok | nc -w 2 "$h" 12181
  echo srvr | nc -w 2 "$h" 12181 | grep Mode
done

Kafka：

/opt/kafka/bin/kafka-broker-api-versions.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka1:19092,kafka2:19092,kafka3:19092 \
  --describe

服务：

systemctl status zookeeper
systemctl status kafka
journalctl -u zookeeper -n 100 --no-pager
journalctl -u kafka -n 100 --no-pager