Kafka 4.1.1 三节点 Ubuntu 集群部署指南

概述

本指南详细介绍如何在Ubuntu环境下部署高可用的 Apache Kafka 4.1.1 三节点集群。本方案采用 KRaft 模式（Kafka Raft Metadata），无需 Zookeeper，通过三节点仲裁机制实现高可用性和数据一致性。

本部署方案基于：

JDK 17：使用最新的LTS版本Java运行环境
Kafka 4.1.1：最新的稳定版本，包含性能优化和安全增强
Ubuntu 20.04+：推荐的部署操作系统

为什么选择三节点集群？

高可用性：任意单节点故障不影响服务连续性
数据安全：三副本机制保证数据不丢失
性能优化：负载均衡，提升整体吞吐量
故障自愈：自动故障检测和恢复机制
扩展性强：支持后续横向扩展

为什么选择 KRaft 模式？

架构简化：无需 Zookeeper，减少组件和故障点
性能提升：元数据管理更高效，延迟更低
运维成本降低：减少维护组件，简化部署
未来趋势：Kafka 官方推荐的部署方式
扩展性更好：支持更大规模的集群

集群架构优势

Raft 协议：保证元数据一致性
控制器仲裁：三节点投票机制确保决策正确性
数据副本：每个分区在三个节点上都有副本
自动故障切换：秒级故障检测和恢复

1. 架构设计

1.1 三节点高可用架构

1.2 KRaft 模式组件说明

组件	说明	作用
Broker	Kafka 代理节点	处理生产者/消费者请求，存储和检索消息
Controller	控制器节点	管理集群元数据，协调分区副本
Quorum Controller	控制器仲裁组	三个 Controller 组成仲裁组，通过 Raft 协议保证元数据一致性
Log Segments	日志段	实际的消息数据存储单元

1.3 高可用机制

1.3.1 Raft 仲裁机制

三节点仲裁：任意 2 个节点达成一致即可做出决策
故障容忍：支持 1 个节点故障不影响集群运行
自动选举：故障节点恢复后自动重新加入集群

1.3.2 数据副本机制

三副本：每个分区在三个节点上都有副本
ISR 机制：维护 In-Sync Replicas 列表
Leader 选举：分区 Leader 故障时自动切换

1.3.3 故障恢复机制

秒级检测：快速发现节点故障
自动切换：无需人工干预
数据一致性：保证故障切换过程中数据不丢失

2. 前置条件

2.1 硬件要求

组件	最低配置	推荐配置
CPU	2核	4核及以上
内存	4GB	8GB及以上
存储	100GB	500GB SSD
网络	1Gbps	10Gbps

2.2 软件要求

Ubuntu 20.04 LTS 或更高版本
Java 17 LTS（推荐）或 Java 11
时间同步服务（chrony 或 ntp）
Kafka 4.1.1

2.3 网络规划

节点	IP 地址	主机名	角色
节点1	192.168.192.211	ubuntu-kafka-node1	Broker + Controller
节点2	192.168.192.212	ubuntu-kafka-node2	Broker + Controller
节点3	192.168.192.213	ubuntu-kafka-node3	Broker + Controller

2.4 端口规划

端口	用途	协议	说明
9092	客户端通信	TCP	PLAINTEXT
9093	控制器通信	TCP	CONTROLLER
9094	SSL通信	TCP	可选
9999	JMX监控	TCP	监控端口

3. 手动部署指南

3.1 环境准备

3.1.1 系统初始化（所有节点）

# 更新系统
sudo apt update && sudo apt upgrade -y

# 安装必要工具
sudo apt install -y wget curl vim git net-tools iputils-ping

# 创建 kafka 用户
sudo useradd -m -s /bin/bash kafka

# 创建相关目录
sudo mkdir -p /opt/kafka
sudo mkdir -p /data/kafka/data
sudo mkdir -p /var/log/kafka

# 设置目录权限
sudo chown -R kafka:kafka /opt/kafka
sudo chown -R kafka:kafka /data/kafka
sudo chown -R kafka:kafka /var/log/kafka

3.1.2 配置时间同步

# 安装 chrony
sudo apt install -y chrony

# 配置时间服务器
sudo tee /etc/chrony/chrony.conf > /dev/null <<EOF
server time1.aliyun.com iburst
server time2.aliyun.com iburst
server ntp.aliyun.com iburst
makestep 1.0 3
rtcsync
EOF

# 重启 chrony 服务
sudo systemctl restart chronyd
sudo systemctl enable chronyd

# 验证时间同步
chronyc sources

3.1.3 配置防火墙

# Ubuntu/Debian (ufw) 防火墙配置
# 开放 Kafka 端口
sudo ufw allow 9092/tcp  # PLAINTEXT
sudo ufw allow 9093/tcp  # CONTROLLER
sudo ufw allow 9094/tcp  # SSL（可选）
sudo ufw allow 9999/tcp  # JMX

# 启用防火墙
sudo ufw --force enable

# 验证规则
sudo ufw status

3.1.4 配置主机名解析

在每个节点上编辑 /etc/hosts：

sudo tee -a /etc/hosts > /dev/null <<EOF
192.168.192.211 ubuntu-kafka-node1
192.168.192.212 ubuntu-kafka-node2
192.168.192.213 ubuntu-kafka-node3
EOF

3.2 安装 Java

# 安装 OpenJDK 17（推荐）
sudo apt install -y openjdk-17-jdk

# 或者安装 OpenJDK 11（兼容）
# sudo apt install -y openjdk-11-jdk

# 验证安装
java -version

# 设置 JAVA_HOME（推荐使用JDK 17）
echo 'export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64' >> ~/.bashrc
echo 'export PATH=$JAVA_HOME/bin:$PATH' >> ~/.bashrc
source ~/.bashrc

3.3 安装 Kafka

在每个节点上执行以下操作：

# 下载 Kafka
cd /opt

wget https://downloads.apache.org/kafka/4.1.1/kafka_2.13-4.1.1.tgz

# 解压
tar -zxvf kafka_2.13-4.1.1.tgz -C /opt/kafka --strip-components=1

# 设置目录权限
sudo chown -R kafka:kafka /opt/kafka

3.4 配置 Kafka

3.4.1 节点1配置（ubuntu-kafka-node1）

创建 config/kraft/server.properties：

# 创建必要的目录
mkdir -p /opt/kafka/config/kraft

cat > /opt/kafka/config/kraft/server.properties << 'EOF'
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

############################# 服务器基础配置 #############################

# 服务器的角色设置，设置此参数使我们进入KRaft模式
process.roles=broker,controller

# 与此实例角色关联的节点ID
node.id=1

# 用于连接到控制器集群的控制器端点列表
controller.quorum.voters=1@ubuntu-kafka-node1:9093,2@ubuntu-kafka-node2:9093,3@ubuntu-kafka-node3:9093

############################# Socket服务器设置 #############################

# Socket服务器监听的地址
# 组合节点（即具有`process.roles=broker,controller`的节点）必须在此处至少列出控制器监听器
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# 用于代理间通信的监听器名称
inter.broker.listener.name=PLAINTEXT

# 代理或控制器将向客户端通告的监听器名称、主机名和端口
advertised.listeners=PLAINTEXT://192.168.192.211:9092

# 控制器使用的监听器名称列表
controller.listener.names=CONTROLLER

# 将监听器名称映射到安全协议，默认情况下它们是相同的
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 服务器用于接收网络请求和发送响应的线程数
num.network.threads=8

# 服务器用于处理请求的线程数，可能包括磁盘I/O
num.io.threads=16

# Socket服务器使用的发送缓冲区(SO_SNDBUF)
socket.send.buffer.bytes=102400

# Socket服务器使用的接收缓冲区(SO_RCVBUF)
socket.receive.buffer.bytes=102400

# Socket服务器将接受的请求的最大大小（防止OOM保护）
socket.request.max.bytes=104857600

############################# 日志基础配置 #############################

# 存储日志文件的目录列表
log.dirs=/data/kafka/data

# 每个主题的默认日志分区数。更多的分区允许更大的消费并行性
num.partitions=6

# 每个数据目录用于启动时日志恢复和关闭时刷新的线程数
num.recovery.threads.per.data.dir=4

############################# 内部主题设置  #############################

# 组元数据内部主题"__consumer_offsets"、"__share_group_state"和"__transaction_state"的复制因子
offsets.topic.replication.factor=3
share.coordinator.state.topic.replication.factor=3
share.coordinator.state.topic.min.isr=2
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

# 控制器引导服务器列表
controller.quorum.bootstrap.servers=ubuntu-kafka-node1:9093,ubuntu-kafka-node2:9093,ubuntu-kafka-node3:9093

############################# 日志刷新策略 #############################

# 消息立即写入文件系统，但默认情况下我们只懒惰地fsync()同步OS缓存

# 在强制将数据刷新到磁盘之前接受的消息数
log.flush.interval.messages=10000

# 消息在日志中停留的最长时间，之后我们强制刷新
log.flush.interval.ms=1000

############################# 日志保留策略 #############################

# 以下配置控制日志段的处理。策略可以设置为在一段时间后或累积到给定大小后删除段

# 有资格因年龄而被删除的日志文件的最小年龄
log.retention.hours=168

# 日志段文件的最大大小。达到此大小时将创建新的日志段
log.segment.bytes=1073741824

# 检查日志段是否可以根据保留策略删除的间隔
log.retention.check.interval.ms=300000

# 自动创建主题
auto.create.topics.enable=true

# 删除主题
delete.topic.enable=true

# 组协调器配置
group.initial.rebalance.delay.ms=0

# JMX 端口
JMX_PORT=9999
EOF

3.4.2 节点2配置（ubuntu-kafka-node2）

创建配置文件：

# 创建必要的目录
mkdir -p /opt/kafka/config/kraft

cat > /opt/kafka/config/kraft/server.properties << 'EOF'
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

############################# 服务器基础配置 #############################

# 服务器的角色设置，设置此参数使我们进入KRaft模式
process.roles=broker,controller

# 与此实例角色关联的节点ID
node.id=2

# 用于连接到控制器集群的控制器端点列表
controller.quorum.voters=1@ubuntu-kafka-node1:9093,2@ubuntu-kafka-node2:9093,3@ubuntu-kafka-node3:9093

############################# Socket服务器设置 #############################

# Socket服务器监听的地址
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# 用于代理间通信的监听器名称
inter.broker.listener.name=PLAINTEXT

# 代理或控制器将向客户端通告的监听器名称、主机名和端口
advertised.listeners=PLAINTEXT://192.168.192.212:9092

# 控制器使用的监听器名称列表
controller.listener.names=CONTROLLER

# 将监听器名称映射到安全协议，默认情况下它们是相同的
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 服务器用于接收网络请求和发送响应的线程数
num.network.threads=8

# 服务器用于处理请求的线程数，可能包括磁盘I/O
num.io.threads=16

# Socket服务器使用的发送缓冲区(SO_SNDBUF)
socket.send.buffer.bytes=102400

# Socket服务器使用的接收缓冲区(SO_RCVBUF)
socket.receive.buffer.bytes=102400

# Socket服务器将接受的请求的最大大小（防止OOM保护）
socket.request.max.bytes=104857600

############################# 日志基础配置 #############################

# 存储日志文件的目录列表
log.dirs=/data/kafka/data

# 每个主题的默认日志分区数
num.partitions=6

# 每个数据目录用于启动时日志恢复和关闭时刷新的线程数
num.recovery.threads.per.data.dir=4

############################# 内部主题设置  #############################

# 组元数据内部主题的复制因子
offsets.topic.replication.factor=3
share.coordinator.state.topic.replication.factor=3
share.coordinator.state.topic.min.isr=2
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

# 控制器引导服务器列表
controller.quorum.bootstrap.servers=ubuntu-kafka-node1:9093,ubuntu-kafka-node2:9093,ubuntu-kafka-node3:9093

############################# 日志刷新策略 #############################

# 在强制将数据刷新到磁盘之前接受的消息数
log.flush.interval.messages=10000

# 消息在日志中停留的最长时间，之后我们强制刷新
log.flush.interval.ms=1000

############################# 日志保留策略 #############################

# 有资格因年龄而被删除的日志文件的最小年龄
log.retention.hours=168

# 日志段文件的最大大小
log.segment.bytes=1073741824

# 检查日志段是否可以根据保留策略删除的间隔
log.retention.check.interval.ms=300000

# 自动创建主题
auto.create.topics.enable=true

# 删除主题
delete.topic.enable=true

# 组协调器配置
group.initial.rebalance.delay.ms=0

# JMX 端口
JMX_PORT=9999
EOF

3.4.3 节点3配置（ubuntu-kafka-node3）

创建配置文件：

# 创建必要的目录
mkdir -p /opt/kafka/config/kraft

cat > /opt/kafka/config/kraft/server.properties << 'EOF'
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

############################# 服务器基础配置 #############################

# 服务器的角色设置，设置此参数使我们进入KRaft模式
process.roles=broker,controller

# 与此实例角色关联的节点ID
node.id=3

# 用于连接到控制器集群的控制器端点列表
controller.quorum.voters=1@ubuntu-kafka-node1:9093,2@ubuntu-kafka-node2:9093,3@ubuntu-kafka-node3:9093

############################# Socket服务器设置 #############################

# Socket服务器监听的地址
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# 用于代理间通信的监听器名称
inter.broker.listener.name=PLAINTEXT

# 代理或控制器将向客户端通告的监听器名称、主机名和端口
advertised.listeners=PLAINTEXT://192.168.192.213:9092

# 控制器使用的监听器名称列表
controller.listener.names=CONTROLLER

# 将监听器名称映射到安全协议，默认情况下它们是相同的
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 服务器用于接收网络请求和发送响应的线程数
num.network.threads=8

# 服务器用于处理请求的线程数，可能包括磁盘I/O
num.io.threads=16

# Socket服务器使用的发送缓冲区(SO_SNDBUF)
socket.send.buffer.bytes=102400

# Socket服务器使用的接收缓冲区(SO_RCVBUF)
socket.receive.buffer.bytes=102400

# Socket服务器将接受的请求的最大大小（防止OOM保护）
socket.request.max.bytes=104857600

############################# 日志基础配置 #############################

# 存储日志文件的目录列表
log.dirs=/data/kafka/data

# 每个主题的默认日志分区数
num.partitions=6

# 每个数据目录用于启动时日志恢复和关闭时刷新的线程数
num.recovery.threads.per.data.dir=4

############################# 内部主题设置  #############################

# 组元数据内部主题的复制因子
offsets.topic.replication.factor=3
share.coordinator.state.topic.replication.factor=3
share.coordinator.state.topic.min.isr=2
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

# 控制器引导服务器列表
controller.quorum.bootstrap.servers=ubuntu-kafka-node1:9093,ubuntu-kafka-node2:9093,ubuntu-kafka-node3:9093

############################# 日志刷新策略 #############################

# 在强制将数据刷新到磁盘之前接受的消息数
log.flush.interval.messages=10000

# 消息在日志中停留的最长时间，之后我们强制刷新
log.flush.interval.ms=1000

############################# 日志保留策略 #############################

# 有资格因年龄而被删除的日志文件的最小年龄
log.retention.hours=168

# 日志段文件的最大大小
log.segment.bytes=1073741824

# 检查日志段是否可以根据保留策略删除的间隔
log.retention.check.interval.ms=300000

# 自动创建主题
auto.create.topics.enable=true

# 删除主题
delete.topic.enable=true

# 组协调器配置
group.initial.rebalance.delay.ms=0

# JMX 端口
JMX_PORT=9999
EOF

3.5 创建系统服务

创建 systemd 服务文件 /etc/systemd/system/kafka.service：

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

[Service]
Type=simple
User=kafka
Group=kafka
Environment="JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64"
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

3.6 集群初始化

3.6.1 生成集群 ID

在任意一个节点上执行：

# 生成集群 ID
cd /opt/kafka
./bin/kafka-storage.sh random-uuid

输出示例：

Slj9VqO5Q8mqVgamNQ5CHw

3.6.2 格式化存储目录

在所有节点上执行格式化命令：

# 格式化存储目录
./bin/kafka-storage.sh format -t Slj9VqO5Q8mqVgamNQ5CHw -c /opt/kafka/config/kraft/server.properties

3.5.3 启动 Kafka 服务

在所有节点上启动 Kafka 服务：

# 重新加载 systemd 配置
sudo systemctl daemon-reload

# 启用开机自启
sudo systemctl enable kafka

# 启动 Kafka 服务
sudo systemctl start kafka

# 检查服务状态
sudo systemctl status kafka

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

3.7 集群验证

3.7.1 检查集群状态

# 查看 Broker 列表
./bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status

# 查看控制器状态
./bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --replication

# 查看集群元数据
./bin/kafka-metadata-shell.sh --snapshot /data/kafka/data/__cluster_metadata-0/00000000000000000000.log

3.4.2 创建测试主题

# 创建 6 分区、3 副本的主题
./bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --partitions 6 \
  --replication-factor 3

# 查看主题详情
./bin/kafka-topics.sh --describe \
  --bootstrap-server localhost:9092 \
  --topic test-topic

输出示例：

Topic: test-topic    TopicId: _TTez0XvRTT6d7qeo76GuGQ
PartitionCount: 6    ReplicationFactor: 3
Configs: segment.bytes=1073741824

    Topic: test-topic    Partition: 0    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3
    Topic: test-topic    Partition: 1    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
    Topic: test-topic    Partition: 2    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2
    Topic: test-topic    Partition: 3    Leader: 1    Replicas: 1,3,2    Isr: 1,3,2
    Topic: test-topic    Partition: 4    Leader: 2    Replicas: 2,1,3    Isr: 2,1,3
    Topic: test-topic    Partition: 5    Leader: 3    Replicas: 3,2,1    Isr: 3,2,1

3.4.3 生产/消费消息测试

生产消息：

# 启动生产者
./bin/kafka-console-producer.sh \
  --bootstrap-server 192.168.192.211:9092 \
  --topic test-topic

# 输入消息（按 Enter 发送）
Hello Kafka Cluster!
This is a test message for three-node cluster.

消费消息：

# 启动消费者（可在任意节点）
./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.192.212:9092 \
  --topic test-topic \
  --from-beginning

3.4.4 高可用性验证

节点故障测试：

# 1. 停止节点 1
sudo systemctl stop kafka

# 2. 查看集群状态（应显示 2 个节点在线）
./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.212:9092 describe --status

# 3. 查看主题状态（Leader 应该自动切换）
./bin/kafka-topics.sh --describe \
  --bootstrap-server 192.168.192.212:9092 \
  --topic test-topic

# 4. 验证消息生产和消费
./bin/kafka-console-producer.sh \
  --bootstrap-server 192.168.192.212:9092 \
  --topic test-topic

# 5. 恢复节点 1
sudo systemctl start kafka

# 6. 等待节点重新加入集群
sleep 30

# 7. 查看主题状态（节点 1 应该重新加入 ISR）
./bin/kafka-topics.sh --describe \
  --bootstrap-server 192.168.192.211:9092 \
  --topic test-topic

数据一致性验证：

# 在不同节点上消费消息，验证数据一致性
./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.192.211:9092 \
  --topic test-topic \
  --from-beginning

./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.192.212:9092 \
  --topic test-topic \
  --from-beginning

./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.192.213:9092 \
  --topic test-topic \
  --from-beginning

3.5 集群健康检查

创建健康检查脚本 /opt/kafka/health-check.sh：

#!/bin/bash

# Kafka 集群健康检查脚本

BOOTSTRAP_SERVER="192.168.192.211:9092,192.168.192.212:9092,192.168.192.213:9092"
KAFKA_HOME="/opt/kafka"

# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

echo "=========================================="
echo "Kafka 三节点集群健康检查"
echo "=========================================="

# 检查 Broker 状态
echo "检查 Broker 状态..."
BROKER_COUNT=$($KAFKA_HOME/bin/kafka-broker-api-versions.sh --bootstrap-server $BOOTSTRAP_SERVER 2>/dev/null | grep -c "id:")
if [ $BROKER_COUNT -eq 3 ]; then
    echo -e "${GREEN}✓${NC} 所有 3 个 Broker 正常运行"
else
    echo -e "${RED}✗${NC} 只有 $BROKER_COUNT 个 Broker 在线"
fi

# 检查 Controller 状态
echo "检查 Controller 状态..."
CONTROLLER_STATUS=$($KAFKA_HOME/bin/kafka-metadata-quorum.sh --bootstrap-server $BOOTSTRAP_SERVER describe --status 2>/dev/null)
if echo "$CONTROLLER_STATUS" | grep -q "LeaderId"; then
    echo -e "${GREEN}✓${NC} Controller 正常"
else
    echo -e "${RED}✗${NC} Controller 异常"
fi

# 检查 Topic 状态
echo "检查 Topic 状态..."
TOPICS=$($KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --list 2>/dev/null)
if [ $? -eq 0 ]; then
    echo -e "${GREEN}✓${NC} Topic 列表获取成功"
    echo "Topic 数量: $(echo "$TOPICS" | wc -l)"
else
    echo -e "${RED}✗${NC} Topic 列表获取失败"
fi

# 检查磁盘空间
echo "检查磁盘空间..."
for i in {1..3}; do
    DISK_USAGE=$(ssh kafka-$i "df -h /data/kafka | awk 'NR==2 {print \$5}' | sed 's/%//'")
    if [ $DISK_USAGE -lt 80 ]; then
        echo -e "kafka-$i: ${GREEN}✓${NC} 磁盘使用率 ${DISK_USAGE}%"
    else
        echo -e "kafka-$i: ${YELLOW}⚠${NC} 磁盘使用率 ${DISK_USAGE}%"
    fi
done

echo "=========================================="
echo "健康检查完成"
echo "=========================================="

执行健康检查：

chmod +x /opt/kafka/health-check.sh
/opt/kafka/health-check.sh

5. 网络配置

5.1 端口配置

端口	协议	说明	是否必需
9092	TCP	PLAINTEXT 监听器	是
9093	TCP	CONTROLLER 监听器	是（KRaft 模式）
9094	TCP	SSL 监听器	可选
9999	TCP	JMX 监控端口	可选

5.2 监听器配置

5.2.1 PLAINTEXT 监听器

# 基本配置
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://<YOUR_IP>:9092

5.2.2 SSL 监听器

# SSL 配置
listeners=SSL://:9094
advertised.listeners=SSL://<YOUR_IP>:9094
ssl.keystore.location=/path/to/keystore.jks
ssl.keystore.password=your-keystore-password
ssl.key.password=your-key-password

5.2.3 SASL 认证

# SASL/PLAIN 配置
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://<YOUR_IP>:9092
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
security.inter.broker.protocol=SASL_PLAINTEXT

5.3 外部访问配置

5.3.1 跨网络访问

如果 Kafka 集群部署在内网，需要从外网访问：

# 内网监听器
listeners=INTERNAL://:9092,EXTERNAL://:9093
inter.broker.listener.name=INTERNAL

# 公网地址
advertised.listeners=INTERNAL://<INTERNAL_IP>:9092,EXTERNAL://<EXTERNAL_IP>:9093

5.3.2 负载均衡配置

使用负载均衡器时的配置：

# 负载均衡器配置
advertised.listeners=PLAINTEXT://<LOAD_BALANCER_IP>:9092
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT

6.2.1 PLAINTEXT 监听器

# 基本配置
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://<YOUR_IP>:9092

5.2.2 SSL 监听器

# SSL 配置
listeners=PLAINTEXT://:9092,SSL://:9094
advertised.listeners=PLAINTEXT://<YOUR_IP>:9092,SSL://<YOUR_IP>:9094

# SSL 配置参数
ssl.keystore.location=/opt/kafka/config/kafka.server.keystore.jks
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>
ssl.truststore.location=/opt/kafka/config/kafka.server.truststore.jks
ssl.truststore.password=<truststore-password>
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.keystore.type=JKS

5.2.3 SASL 认证

# SASL 配置
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://<YOUR_IP>:9092

# 启用 SASL 认证
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

# JAAS 配置
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin

创建 JAAS 配置文件 config/kafka_server_jaas.conf：

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_user1="user1-secret"
    user_user2="user2-secret";
};

修改启动脚本：

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"

5.3 外部访问配置

5.3.1 跨网络访问

如果 Kafka 集群部署在内网，需要从外网访问：

# 内网监听器
listeners=INTERNAL://:9092,EXTERNAL://:9093
inter.broker.listener.name=INTERNAL

# 公网地址
advertised.listeners=INTERNAL://<INTERNAL_IP>:9092,EXTERNAL://<EXTERNAL_IP>:9093

# 监听器映射
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT

5.3.2 负载均衡配置

使用 HAProxy 或 Nginx 配置负载均衡：

HAProxy 配置示例：

frontend kafka_frontend
    bind *:9092
    mode tcp
    default_backend kafka_backend

backend kafka_backend
    mode tcp
    balance roundrobin
    server kafka1 192.168.192.211:9092 check
    server kafka2 192.168.192.212:9092 check
    server kafka3 192.168.192.213:9092 check

7. 持久化存储配置

7.1 存储目录规划

# 数据目录
/data/kafka/data/
├── __cluster_metadata-0/          # 集群元数据
├── test-topic-0/                  # Topic 分区数据
├── test-topic-1/
└── test-topic-2/

# 日志目录
/var/log/kafka/
├── server.log                     # 服务器日志
├── controller.log                 # 控制器日志
├── state-change.log               # 状态变更日志
└── log-cleaner.log                # 日志清理日志

7.2 日志段配置

# 日志段大小（默认 1GB）
log.segment.bytes=1073741824

# 日志保留时间（默认 7 天）
log.retention.hours=168

# 日志保留大小
log.retention.bytes=10737418240

# 日志检查间隔
log.retention.check.interval.ms=300000

# 日志清理策略
log.cleanup.policy=delete

# 日志压缩策略
log.cleanup.policy=compact
log.cleaner.enable=true
log.cleaner.threads=4

7.3 磁盘 I/O 优化

7.3.1 文件系统选择

推荐使用 XFS 文件系统：

# 创建 XFS 文件系统
mkfs.xfs -f /dev/sdb1

# 挂载
mount -o noatime,nodiratime /dev/sdb1 /data/kafka

# 添加到 /etc/fstab
echo "/dev/sdb1 /data/kafka xfs defaults,noatime,nodiratime 0 0" >> /etc/fstab

7.3.2 磁盘调度器

# 设置磁盘调度器为 deadline 或 noop
echo deadline > /sys/block/sdb/queue/scheduler

# 永久配置
cat >> /etc/rc.local << EOF
echo deadline > /sys/block/sdb/queue/scheduler
EOF

7.3.3 RAID 配置

生产环境建议使用 RAID 10：

# 创建 RAID 10
mdadm --create /dev/md0 --level=10 --raid-devices=4 /dev/sdb1 /dev/sdc1 /dev/sdd1 /dev/sde1

# 创建文件系统
mkfs.xfs -f /dev/md0

# 挂载
mount -o noatime,nodiratime /dev/md0 /data/kafka

8. 监控与告警

8.1 JMX 监控配置

8.1.1 启用 JMX 端口

在 server.properties 中配置：

# JMX 端口
JMX_PORT=9999

修改启动脚本：

export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false \
  -Djava.rmi.server.hostname=<YOUR_IP> \
  -Dcom.sun.management.jmxremote.rmi.port=9999"

8.1.2 验证 JMX 连接

# 使用 jconsole 连接
jconsole <YOUR_IP>:9999

# 使用 jcmd 查看
jcmd <PID> VM.version

8.2 Prometheus 监控

8.2.1 部署 JMX Exporter

下载 JMX Exporter：

cd /opt/kafka
wget https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.20.0/jmx_prometheus_javaagent-0.20.0.jar

创建 JMX Exporter 配置文件 config/jmx-exporter.yml：

startDelaySeconds: 0
hostPort: 127.0.0.1:9999
jmxUrl: service:jmx:rmi:///jndi/rmi://127.0.0.1:9999/jmxrmi
ssl: false
lowercaseOutputName: true
lowercaseOutputLabelNames: true

rules:
  - pattern: 'kafka.server<type=(.+), name=(.+)PerSec\w*><>Count'
    name: kafka_server_$1_$2_total
    type: COUNTER

  - pattern: 'kafka.server<type=(.+), name=(.+)PerSec\w*><>OneMinuteRate'
    name: kafka_server_$1_$2_rate_1m
    type: GAUGE

  - pattern: 'kafka.server<type=(.+), name=(.+)PerSec\w*><>FiveMinuteRate'
    name: kafka_server_$1_$2_rate_5m
    type: GAUGE

  - pattern: 'kafka.server<type=(.+), name=(.+)PerSec\w*><>FifteenMinuteRate'
    name: kafka_server_$1_$2_rate_15m
    type: GAUGE

  - pattern: 'kafka.server<type=(.+), name=(.+)><>Value'
    name: kafka_server_$1_$2
    type: GAUGE

  - pattern: 'kafka.log<type=Log, name=(.+), topic=(.+), partition=(.+)><>Value'
    name: kafka_log_$1
    labels:
      topic: "$2"
      partition: "$3"
    type: GAUGE

  - pattern: 'kafka.controller<type=(.+), name=(.+)><>Value'
    name: kafka_controller_$1_$2
    type: GAUGE

  - pattern: 'kafka.network<type=(.+), name=(.+)><>Value'
    name: kafka_network_$1_$2
    type: GAUGE

  - pattern: 'java.lang<type=Memory><>HeapMemoryUsage'
    name: jvm_memory_heap_bytes
    type: GAUGE

  - pattern: 'java.lang<type=Memory><>NonHeapMemoryUsage'
    name: jvm_memory_nonheap_bytes
    type: GAUGE

  - pattern: 'java.lang<type=GarbageCollector, name=(.+)><>CollectionCount'
    name: jvm_gc_collection_count_total
    labels:
      gc: "$1"
    type: COUNTER

  - pattern: 'java.lang<type=GarbageCollector, name=(.+)><>CollectionTime'
    name: jvm_gc_collection_time_seconds_total
    labels:
      gc: "$1"
    type: COUNTER

修改 Kafka 启动脚本：

export KAFKA_OPTS="-javaagent:/opt/kafka/jmx_prometheus_javaagent-0.20.0.jar=9998:/opt/kafka/config/jmx-exporter.yml"

8.2.2 配置 Prometheus 抓取

在 Prometheus 配置文件中添加：

scrape_configs:
  - job_name: 'kafka'
    static_configs:
      - targets:
          - '192.168.192.211:9998'
          - '192.168.192.212:9998'
          - '192.168.192.213:9998'
    scrape_interval: 15s
    scrape_timeout: 10s

8.2.3 Grafana 仪表盘

导入 Kafka 监控仪表盘（推荐使用 Grafana 官方仪表盘）：

Kafka Exporter: ID 7589
JMX Exporter: ID 721

8.3 告警规则

创建 Prometheus 告警规则 kafka-alerts.yml：

groups:
  - name: kafka
    interval: 30s
    rules:
      - alert: KafkaBrokerDown
        expr: up{job="kafka"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Kafka Broker down"
          description: "Kafka Broker {{ $labels.instance }} is down"

      - alert: KafkaUnderReplicatedPartitions
        expr: kafka_server_ReplicaManager_UnderReplicatedPartitions > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Kafka under replicated partitions"
          description: "Kafka cluster has {{ $value }} under replicated partitions"

      - alert: KafkaOfflinePartitionsCount
        expr: kafka_controller_OfflinePartitionsCount > 0
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Kafka offline partitions"
          description: "Kafka cluster has {{ $value }} offline partitions"

      - alert: KafkaControllerActiveCount
        expr: kafka_controller_KafkaController_ActiveControllerCount < 1
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Kafka controller down"
          description: "Kafka controller is not active"

      - alert: KafkaRequestHandlerAvgIdlePercent
        expr: kafka_server_KafkaRequestHandlerPool_AvgIdlePercent < 20
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Kafka request handler busy"
          description: "Kafka request handler average idle percent is {{ $value }}%"

      - alert: KafkaLogFlushRateAndTimeMs
        expr: rate(kafka_log_LogFlushStats_LogFlushRateAndTimeMs_count[5m]) == 0
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Kafka log flush rate zero"
          description: "Kafka log flush rate is zero for 10 minutes"

      - alert: KafkaDiskUsageHigh
        expr: (1 - (node_filesystem_avail_bytes{mountpoint="/data/kafka"} / node_filesystem_size_bytes{mountpoint="/data/kafka"})) * 100 > 80
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Kafka disk usage high"
          description: "Kafka disk usage is {{ $value }}%"

      - alert: KafkaDiskUsageCritical
        expr: (1 - (node_filesystem_avail_bytes{mountpoint="/data/kafka"} / node_filesystem_size_bytes{mountpoint="/data/kafka"})) * 100 > 90
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Kafka disk usage critical"
          description: "Kafka disk usage is {{ $value }}%"

      - alert: KafkaConsumerLag
        expr: kafka_consumergroup_lag > 10000
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Kafka consumer lag high"
          description: "Kafka consumer group {{ $labels.consumergroup }} lag is {{ $value }}"

      - alert: KafkaMessageInRateLow
        expr: rate(kafka_server_BrokerTopicMetrics_BytesInPerSec[5m]) < 100
        for: 30m
        labels:
          severity: info
        annotations:
          summary: "Kafka message in rate low"
          description: "Kafka message in rate is {{ $value }} bytes/sec"

9. 运维操作

9.1 日常运维

9.1.1 Topic 管理

创建 Topic：

# 创建 Topic
./bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --partitions 6 \
  --replication-factor 3

# 创建带配置的 Topic
./bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --partitions 6 \
  --replication-factor 3 \
  --config retention.ms=86400000 \
  --config segment.bytes=1073741824

查看 Topic：

# 列出所有 Topic
./bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# 查看 Topic 详情
./bin/kafka-topics.sh --describe \
  --bootstrap-server localhost:9092 \
  --topic my-topic

# 查看 Topic 配置
./bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --describe

修改 Topic：

# 增加分区
./bin/kafka-topics.sh --alter \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --partitions 12

# 修改配置
./bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --alter \
  --add-config retention.ms=172800000

# 删除配置
./bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name my-topic \
  --alter \
  --delete-config retention.ms

删除 Topic：

./bin/kafka-topics.sh --delete \
  --bootstrap-server localhost:9092 \
  --topic my-topic

9.1.2 消费组管理

查看消费组：

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

# 查看消费组详情
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group \
  --describe

# 查看消费组状态
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group \
  --state

重置消费组偏移量：

# 重置到最早
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group \
  --reset-offsets \
  --to-earliest \
  --topic my-topic \
  --execute

# 重置到最新
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group \
  --reset-offsets \
  --to-latest \
  --topic my-topic \
  --execute

# 重置到指定时间戳
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group \
  --reset-offsets \
  --to-datetime 2024-01-01T00:00:00.000 \
  --topic my-topic \
  --execute

删除消费组：

./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group \
  --delete

9.1.3 日志清理

手动清理日志：

# 清理指定 Topic 的日志
./bin/kafka-delete-records.sh --bootstrap-server localhost:9092 \
  --offset-json-file offsets.json

offsets.json 示例：

{
  "partitions": [
    {
      "topic": "my-topic",
      "partition": 0,
      "offset": 1000
    },
    {
      "topic": "my-topic",
      "partition": 1,
      "offset": 1000
    }
  ]
}

查看日志段：

# 查看 Topic 分区日志
ls -lh /data/kafka/data/my-topic-0/

# 查看日志段信息
./bin/kafka-dump-log.sh --files /data/kafka/data/my-topic-0/00000000000000000000.log

9.2 扩缩容

9.2.1 增加 Broker 节点

在新节点上安装 Kafka（参考生产环境部署章节）
配置新节点的 server.properties
启动新节点
重平衡分区

# 重平衡分区
./bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file reassign.json \
  --execute

# 查看重平衡状态
./bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file reassign.json \
  --verify

reassign.json 示例：

{
  "version": 1,
  "partitions": [
    {
      "topic": "my-topic",
      "partition": 0,
      "replicas": [1, 2, 4]
    },
    {
      "topic": "my-topic",
      "partition": 1,
      "replicas": [2, 3, 4]
    }
  ]
}

9.2.2 减少 Broker 节点

将节点上的分区迁移到其他节点
等待迁移完成
停止节点

# 生成迁移计划
./bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file topics-to-move.json \
  --broker-list "1,2,3" \
  --generate

# 执行迁移
./bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file expand-cluster-reassignment.json \
  --execute

# 验证迁移
./bin/kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file expand-cluster-reassignment.json \
  --verify

9.2.3 重平衡分区

使用 Cruise Control 自动重平衡：

# 生成重平衡计划
curl -X POST "http://localhost:8090/kafkacruisecontrol/rebalance?dryrun=true" \
  -H "Content-Type: application/json" \
  -d '{
    "goals": ["RackAwareGoal", "ReplicaCapacityGoal", "DiskCapacityGoal"],
    "allow_capacity_estimation": true
  }'

# 执行重平衡
curl -X POST "http://localhost:8090/kafkacruisecontrol/rebalance" \
  -H "Content-Type: application/json" \
  -d '{
    "goals": ["RackAwareGoal", "ReplicaCapacityGoal", "DiskCapacityGoal"],
    "allow_capacity_estimation": true
  }'

9.3 升级与回滚

9.3.1 Kafka 版本升级

滚动升级步骤：

下载新版本 Kafka
逐个节点升级
验证集群状态

# 1. 停止节点
sudo systemctl stop kafka

# 2. 备份配置
cp /opt/kafka/config/kraft/server.properties /opt/kafka/config/kraft/server.properties.bak

# 3. 下载新版本
cd /opt/kafka
wget https://downloads.apache.org/kafka/3.8.0/kafka_2.13-3.8.0.tgz
tar -xzf kafka_2.13-3.8.0.tgz

# 4. 更新符号链接
rm kafka
ln -s kafka_2.13-3.8.0 kafka

# 5. 启动节点
sudo systemctl start kafka

# 6. 验证
./bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status

9.3.2 配置变更

# 1. 修改配置
vi /opt/kafka/config/kraft/server.properties

# 2. 动态更新配置（支持的配置）
./bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type brokers \
  --entity-name 1 \
  --alter \
  --add-config log.retention.hours=168

# 3. 重启节点（不支持的配置）
sudo systemctl restart kafka

9.3.3 紧急回滚

# 1. 停止节点
sudo systemctl stop kafka

# 2. 恢复旧版本
rm /opt/kafka
ln -s kafka_2.13-4.1.1 kafka

# 3. 恢复配置
cp /opt/kafka/config/kraft/server.properties.bak /opt/kafka/config/kraft/server.properties

# 4. 启动节点
sudo systemctl start kafka

9.4 备份与恢复

9.4.1 数据备份策略

完整备份：

# 停止 Kafka
sudo systemctl stop kafka

# 备份数据目录
tar -czf /backup/kafka/kafka_backup_$(date +%Y%m%d_%H%M%S).tar.gz /data/kafka/data

# 备份配置文件
tar -czf /backup/kafka/kafka_config_$(date +%Y%m%d_%H%M%S).tar.gz /opt/kafka/config

# 启动 Kafka
sudo systemctl start kafka

增量备份：

# 使用 rsync 进行增量备份
rsync -avz --delete /data/kafka/data/ /backup/kafka/data/

9.4.2 灾难恢复方案

从备份恢复：

# 1. 停止 Kafka
sudo systemctl stop kafka

# 2. 清空数据目录
rm -rf /data/kafka/data/*

# 3. 恢复数据
tar -xzf /backup/kafka/kafka_backup_20240101_120000.tar.gz -C /

# 4. 恢复配置
tar -xzf /backup/kafka/kafka_config_20240101_120000.tar.gz -C /

# 5. 启动 Kafka
sudo systemctl start kafka

跨数据中心复制：

使用 MirrorMaker 2.0 进行跨数据中心复制：

# 配置 MirrorMaker 2.0
./bin/connect-mirror-maker.sh mm2.properties

mm2.properties 示例：

# 集群配置
clusters = primary, backup

primary.bootstrap.servers = primary-kafka1:9092,primary-kafka2:9092,primary-kafka3:9092
backup.bootstrap.servers = backup-kafka1:9092,backup-kafka2:9092,backup-kafka3:9092

# 复制配置
primary->backup.enabled = true
backup->primary.enabled = false

primary->backup.topics = .*
backup->primary.topics = .*

# 主题配置
replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy

10. 安全配置

10.1 SASL 认证

10.1.1 PLAIN 认证

服务器配置：

创建 JAAS 配置文件 config/kafka_server_jaas.conf：

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_user1="user1-secret"
    user_user2="user2-secret";
};

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret";
};

修改 server.properties：

# 启用 SASL
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

# ACL 配置
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin

修改启动脚本：

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"

客户端配置：

# 创建客户端 JAAS 配置
cat > /tmp/kafka_client_jaas.conf << EOF
KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="user1"
    password="user1-secret";
};
EOF

# 设置环境变量
export KAFKA_OPTS="-Djava.security.auth.login.config=/tmp/kafka_client_jaas.conf"

# 使用 SASL 连接
./bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --producer.config /tmp/producer.properties

producer.properties 示例：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="user1" password="user1-secret";

10.1.2 SCRAM 认证

创建用户：

# 创建 SCRAM 用户
./bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'SCRAM-SHA-256=[password=user1-secret]' \
  --entity-type users \
  --entity-name user1

# 创建管理员用户
./bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'SCRAM-SHA-256=[password=admin-secret]' \
  --entity-type users \
  --entity-name admin

修改 server.properties：

listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256

客户端配置：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="user1" password="user1-secret";

10.2 SSL 加密

10.2.1 证书生成

生成 CA 证书：

# 生成 CA 私钥
openssl genrsa -out ca-key.pem 2048

# 生成 CA 证书
openssl req -new -x509 -key ca-key.pem -out ca-cert.pem -days 3650 \
  -subj "/CN=Kafka CA"

生成服务器证书：

# 生成服务器私钥
openssl genrsa -out server-key.pem 2048

# 生成证书签名请求
openssl req -new -key server-key.pem -out server-csr.pem \
  -subj "/CN=ubuntu-kafka-node1" \
  -addext "subjectAltName=DNS:ubuntu-kafka-node1,DNS:localhost,IP:192.168.192.211"

# 使用 CA 签名
openssl x509 -req -CA ca-cert.pem -CAkey ca-key.pem -in server-csr.pem \
  -out server-cert.pem -days 3650 -CAcreateserial

生成客户端证书：

# 生成客户端私钥
openssl genrsa -out client-key.pem 2048

# 生成证书签名请求
openssl req -new -key client-key.pem -out client-csr.pem \
  -subj "/CN=kafka-client"

# 使用 CA 签名
openssl x509 -req -CA ca-cert.pem -CAkey ca-key.pem -in client-csr.pem \
  -out client-cert.pem -days 3650 -CAcreateserial

10.2.2 SSL 配置

服务器配置：

# SSL 监听器
listeners=PLAINTEXT://:9092,SSL://:9094
advertised.listeners=PLAINTEXT://192.168.192.211:9092,SSL://192.168.192.211:9094

# SSL 配置
ssl.keystore.location=/opt/kafka/config/server.keystore.jks
ssl.keystore.password=keystore-password
ssl.key.password=key-password
ssl.truststore.location=/opt/kafka/config/server.truststore.jks
ssl.truststore.password=truststore-password
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.keystore.type=JKS

创建 JKS Keystore：

# 转换 PEM 到 PKCS12
openssl pkcs12 -export -in server-cert.pem -inkey server-key.pem \
  -out server.p12 -name kafka-server -CAfile ca-cert.pem -caname root \
  -password pass:keystore-password

# 转换 PKCS12 到 JKS
keytool -importkeystore \
  -deststorepass keystore-password \
  -destkeypass key-password \
  -destkeystore server.keystore.jks \
  -srckeystore server.p12 \
  -srcstoretype PKCS12 \
  -srcstorepass keystore-password \
  -alias kafka-server

# 导入 CA 证书到 Truststore
keytool -import -alias ca-cert \
  -file ca-cert.pem \
  -keystore server.truststore.jks \
  -storepass truststore-password \
  -noprompt

客户端配置：

security.protocol=SSL
ssl.truststore.location=/opt/kafka/config/client.truststore.jks
ssl.truststore.password=truststore-password
ssl.keystore.location=/opt/kafka/config/client.keystore.jks
ssl.keystore.password=keystore-password
ssl.key.password=key-password

11. 性能优化

11.1 JVM 参数调优

11.1.1 堆内存配置

# 堆内存建议为物理内存的 50-70%
# 示例：8GB 内存，堆内存配置为 4-6GB

# 小型环境（2GB 内存）
KAFKA_HEAP_OPTS="-Xms1g -Xmx1g"

# 中型环境（8GB 内存）
KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"

# 大型环境（16GB 内存）
KAFKA_HEAP_OPTS="-Xms8g -Xmx8g"

11.1.2 GC 参数

# G1GC 配置（推荐）
KAFKA_HEAP_OPTS="-Xms4g -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent"

# CMS GC 配置（旧版本）
KAFKA_HEAP_OPTS="-Xms4g -Xmx4g -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70"

11.1.3 其他 JVM 参数

# 完整的 JVM 配置
KAFKA_OPTS="-Xms4g -Xmx4g \
  -XX:+UseG1GC \
  -XX:MaxGCPauseMillis=20 \
  -XX:InitiatingHeapOccupancyPercent=35 \
  -XX:+ExplicitGCInvokesConcurrent \
  -XX:MaxInlineLevel=15 \
  -XX:+UseStringDeduplication \
  -Djava.awt.headless=true \
  -Dcom.sun.management.jmxremote \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false \
  -Djava.rmi.server.hostname=<YOUR_IP> \
  -Dcom.sun.management.jmxremote.rmi.port=9999"

11.2 Kafka 参数调优

11.2.1 网络线程配置

# 网络线程数（建议为 CPU 核心数）
num.network.threads=8

# I/O 线程数（建议为 CPU 核心数的 2 倍）
num.io.threads=16

# 后台线程数
background.threads=10

11.2.2 日志刷盘策略

# 消息刷盘策略
log.flush.interval.messages=10000
log.flush.interval.ms=1000

# 控制器刷盘策略
log.flush.offset.checkpoint.interval.ms=60000
log.flush.start.offset.checkpoint.interval.ms=60000

11.2.3 批处理配置

# 生产者批处理
batch.size=16384
linger.ms=5

# 消费者拉取
fetch.min.bytes=1
fetch.max.wait.ms=500
max.partition.fetch.bytes=1048576

11.2.4 压缩配置

# 压缩类型（none, gzip, snappy, lz4, zstd）
compression.type=lz4

# 压缩级别
compression.lz4.level=9
compression.zstd.level=3

11.3 操作系统优化

11.3.1 文件句柄限制

# 临时修改
ulimit -n 65536

# 永久修改
cat >> /etc/security/limits.conf << EOF
* soft nofile 65536
* hard nofile 65536
kafka soft nofile 131072
kafka hard nofile 131072
EOF

11.3.2 TCP 参数优化

# 修改 /etc/sysctl.conf
cat >> /etc/sysctl.conf << EOF
# TCP 优化
net.core.somaxconn = 1024
net.core.netdev_max_backlog = 5000
net.ipv4.tcp_max_syn_backlog = 1024
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 30
net.ipv4.tcp_keepalive_time = 1200
net.ipv4.tcp_max_tw_buckets = 5000

# 网络缓冲区
net.core.rmem_default = 262144
net.core.rmem_max = 16777216
net.core.wmem_default = 262144
net.core.wmem_max = 16777216
net.ipv4.tcp_rmem = 4096 87380 16777216
net.ipv4.tcp_wmem = 4096 65536 16777216
EOF

# 应用配置
sysctl -p

11.3.3 磁盘调度器

# 设置磁盘调度器为 deadline
echo deadline > /sys/block/sdb/queue/scheduler

# 永久配置
cat >> /etc/rc.local << EOF
echo deadline > /sys/block/sdb/queue/scheduler
EOF

# 或者使用 udev 规则
cat > /etc/udev/rules.d/60-scheduler.rules << EOF
ACTION=="add|change", KERNEL=="sd*", ATTR{queue/scheduler}="deadline"
EOF

11. 高可用性测试与故障排除

11.1 三节点集群高可用性测试

11.1.1 节点故障切换测试

测试场景 1：单节点故障

# 1. 初始状态检查
echo "=== 初始集群状态 ==="
./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.211:9092 describe --status
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.211:9092 --topic test-topic

# 2. 模拟节点 1 故障
echo "=== 模拟 ubuntu-kafka-node1 节点故障 ==="
sudo systemctl stop kafka
sleep 10

# 3. 验证集群状态（应该自动切换到节点 2 为 Leader）
echo "=== 故障后集群状态 ==="
./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.212:9092 describe --status
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.212:9092 --topic test-topic

# 4. 验证数据服务连续性
echo "=== 测试消息生产和消费 ==="
echo "故障测试消息 $(date)" | ./bin/kafka-console-producer.sh \
  --bootstrap-server 192.168.192.212:9092 \
  --topic test-topic

# 5. 恢复故障节点
echo "=== 恢复 ubuntu-kafka-node1 节点 ==="
sudo systemctl start kafka
sleep 30

# 6. 最终状态检查
echo "=== 最终集群状态 ==="
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.211:9092 --topic test-topic
./bin/kafka-console-consumer.sh --bootstrap-server 192.168.192.211:9092 --topic test-topic --from-beginning | tail -5

11.1.2 Controller 选举测试

测试场景 2：Controller 故障切换

# 1. 查看当前 Controller
echo "=== 当前 Controller 状态 ==="
./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.211:9092 describe --status

# 2. 停止当前 Controller 节点
CURRENT_CONTROLLER=$(./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.211:9092 describe --status | grep "LeaderId" | awk '{print $2}')
echo "当前 Controller 节点: $CURRENT_CONTROLLER"

if [ "$CURRENT_CONTROLLER" = "1" ]; then
    sudo systemctl stop kafka  # 在 ubuntu-kafka-node1 上执行
elif [ "$CURRENT_CONTROLLER" = "2" ]; then
    # SSH 到 ubuntu-kafka-node2 执行
    ssh ubuntu-kafka-node2 "sudo systemctl stop kafka"
else
    # SSH 到 ubuntu-kafka-node3 执行
    ssh ubuntu-kafka-node3 "sudo systemctl stop kafka"
fi

# 3. 等待新 Controller 选举
sleep 15

# 4. 验证新 Controller
echo "=== 新 Controller 状态 ==="
./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.212:9092 describe --status

# 5. 恢复原 Controller
if [ "$CURRENT_CONTROLLER" = "1" ]; then
    sudo systemctl start kafka
elif [ "$CURRENT_CONTROLLER" = "2" ]; then
    ssh ubuntu-kafka-node2 "sudo systemctl start kafka"
else
    ssh ubuntu-kafka-node3 "sudo systemctl start kafka"
fi

sleep 30

11.1.3 网络分区测试

测试场景 3：模拟网络分区

# 1. 使用 iptables 模拟网络隔离
echo "=== 模拟网络分区 ==="
# 在 ubuntu-kafka-node1 上隔离网络
sudo iptables -A INPUT -s 192.168.192.212 -j DROP
sudo iptables -A INPUT -s 192.168.192.213 -j DROP
sudo iptables -A OUTPUT -d 192.168.192.212 -j DROP
sudo iptables -A OUTPUT -d 192.168.192.213 -j DROP

# 2. 观察集群行为
echo "=== 观察集群状态变化 ==="
for i in {1..10}; do
    echo "--- 第 $i 次检查 ---"
    ./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.212:9092 describe --status
    sleep 5
done

# 3. 恢复网络
echo "=== 恢复网络连接 ==="
sudo iptables -F

# 4. 验证集群恢复
sleep 30
./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.211:9092 describe --status

11.2 常见故障排除

11.2.1 集群启动失败

问题：三节点集群无法启动

# 1. 检查各节点服务状态
for i in {1..3}; do
    echo "=== kafka-$i 状态 ==="
    ssh kafka-$i "systemctl status kafka"
done

# 2. 检查端口占用
for i in {1..3}; do
    echo "=== kafka-$i 端口 ==="
    ssh kafka-$i "netstat -tlnp | grep -E '9092|9093'"
done

# 3. 检查配置一致性
for i in {1..3}; do
    echo "=== kafka-$i 配置检查 ==="
    ssh kafka-$i "grep 'controller.quorum.voters' /opt/kafka/config/kraft/server.properties"
    ssh kafka-$i "grep 'node.id' /opt/kafka/config/kraft/server.properties"
done

# 4. 检查集群 ID 一致性
for i in {1..3}; do
    echo "=== kafka-$i 集群 ID ==="
    ssh kafka-$i "cat /data/kafka/data/meta.properties"
done

# 5. 查看详细日志
for i in {1..3}; do
    echo "=== kafka-$i 错误日志 ==="
    ssh kafka-$i "tail -20 /opt/kafka/logs/server.log | grep ERROR"
done

11.2.2 数据不一致问题

问题：副本数据不同步

# 1. 检查 ISR 状态
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.211:9092 --topic test-topic

# 2. 检查各节点数据目录
for i in {1..3}; do
    echo "=== kafka-$i 数据目录 ==="
    ssh kafka-$i "ls -la /data/kafka/data/test-topic-* | head -5"
done

# 3. 检查日志段完整性
for i in {1..3}; do
    echo "=== kafka-$i 日志段检查 ==="
    ssh kafka-$i "/opt/kafka/bin/kafka-dump-log.sh --files /data/kafka/data/test-topic-0/00000000000000000000.log"
done

# 4. 强制同步副本
./bin/kafka-leader-election.sh --bootstrap-server 192.168.192.211:9092 \
  --election-type preferred --all-topic-partitions

11.2.3 性能瓶颈分析

问题：集群性能下降

# 1. 集群整体性能监控
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.211:9092 --under-replicated-partitions
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.211:9092 --unavailable-partitions

# 2. 各节点资源使用情况
for i in {1..3}; do
    echo "=== kafka-$i 资源使用 ==="
    ssh kafka-$i "top -bn1 | head -10"
    ssh kafka-$i "iostat -x 1 3"
    ssh kafka-$i "free -h"
done

# 3. JVM 性能分析
for i in {1..3}; do
    PID=$(ssh kafka-$i "pgrep -f kafka.Kafka")
    echo "=== kafka-$i JVM 状态 ==="
    ssh kafka-$i "jstat -gc $PID 1000 3"
done

# 4. 网络性能检查
for i in {1..3}; do
    echo "=== kafka-$i 网络状态 ==="
    ssh kafka-$i "sar -n DEV 1 3"
    ssh kafka-$i "netstat -i"
done

11.3 诊断工具和脚本

11.3.1 集群健康检查脚本

创建 /opt/kafka/cluster-health-check.sh：

#!/bin/bash

# Kafka 三节点集群健康检查脚本

NODES=("192.168.192.211" "192.168.192.212" "192.168.192.213")
BOOTSTRAP_SERVER="192.168.192.211:9092,192.168.192.212:9092,192.168.192.213:9092"
KAFKA_HOME="/opt/kafka"

# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'

check_node() {
    local node=$1
    local node_name=$2
    
    echo "检查 $node_name ($node)..."
    
    # 检查 SSH 连接
    if ssh -o ConnectTimeout=5 $node "echo test" >/dev/null 2>&1; then
        echo -e "  SSH连接: ${GREEN}✓${NC}"
    else
        echo -e "  SSH连接: ${RED}✗${NC}"
        return 1
    fi
    
    # 检查 Kafka 服务
    if ssh $node "systemctl is-active kafka" >/dev/null 2>&1; then
        echo -e "  Kafka服务: ${GREEN}✓${NC}"
    else
        echo -e "  Kafka服务: ${RED}✗${NC}"
        return 1
    fi
    
    # 检查端口
    if ssh $node "netstat -tlnp | grep -q ':9092'"; then
        echo -e "  9092端口: ${GREEN}✓${NC}"
    else
        echo -e "  9092端口: ${RED}✗${NC}"
    fi
    
    # 检查磁盘使用率
    disk_usage=$(ssh $node "df /data/kafka | awk 'NR==2 {print \$5}' | sed 's/%//'")
    if [ "$disk_usage" -lt 80 ]; then
        echo -e "  磁盘使用率: ${GREEN}${disk_usage}%${NC}"
    elif [ "$disk_usage" -lt 90 ]; then
        echo -e "  磁盘使用率: ${YELLOW}${disk_usage}%${NC}"
    else
        echo -e "  磁盘使用率: ${RED}${disk_usage}%${NC}"
    fi
    
    return 0
}

echo "=========================================="
echo "Kafka 三节点集群健康检查报告"
echo "检查时间: $(date)"
echo "=========================================="

# 检查各节点状态
node_count=0
for i in "${!NODES[@]}"; do
    node_name="kafka-$((i+1))"
    if check_node "${NODES[$i]}" "$node_name"; then
        ((node_count++))
    fi
    echo ""
done

echo "节点健康状态: $node_count/3 正常"
echo "=========================================="

# 集群级别检查
if [ $node_count -ge 2 ]; then
    echo "检查集群元数据..."
    if $KAFKA_HOME/bin/kafka-metadata-quorum.sh --bootstrap-server $BOOTSTRAP_SERVER describe --status >/dev/null 2>&1; then
        echo -e "集群元数据: ${GREEN}✓${NC}"
    else
        echo -e "集群元数据: ${RED}✗${NC}"
    fi
    
    echo "检查控制器状态..."
    controller_status=$($KAFKA_HOME/bin/kafka-metadata-quorum.sh --bootstrap-server $BOOTSTRAP_SERVER describe --status 2>/dev/null)
    if echo "$controller_status" | grep -q "LeaderId"; then
        leader_id=$(echo "$controller_status" | grep "LeaderId" | awk '{print $2}')
        echo -e "控制器Leader: ${GREEN}节点 $leader_id${NC}"
    else
        echo -e "控制器Leader: ${RED}未选举${NC}"
    fi
    
    echo "检查主题状态..."
    under_replicated=$($KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --describe | grep -c "UnderReplicated")
    if [ "$under_replicated" -eq 0 ]; then
        echo -e "副本状态: ${GREEN}全部同步${NC}"
    else
        echo -e "副本状态: ${YELLOW}$under_replicated 个分区未同步${NC}"
    fi
else
    echo -e "${RED}警告: 集群节点不足，无法进行集群级别检查${NC}"
fi

echo "=========================================="
echo "健康检查完成"
echo "=========================================="

使用方法：

chmod +x /opt/kafka/cluster-health-check.sh
/opt/kafka/cluster-health-check.sh

11.3.2 故障自动恢复脚本

创建 /opt/kafka/auto-recovery.sh：

#!/bin/bash

# Kafka 集群自动恢复脚本

BOOTSTRAP_SERVER="192.168.192.211:9092,192.168.192.212:9092,192.168.192.213:9092"
KAFKA_HOME="/opt/kafka"
LOG_FILE="/var/log/kafka/auto-recovery.log"

log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" | tee -a $LOG_FILE
}

# 检查集群健康状态
check_cluster_health() {
    local unhealthy_count=0
    
    # 检查 Broker 数量
    broker_count=$($KAFKA_HOME/bin/kafka-broker-api-versions.sh --bootstrap-server $BOOTSTRAP_SERVER 2>/dev/null | grep -c "id:")
    if [ "$broker_count" -lt 3 ]; then
        log "WARNING: 只有 $broker_count 个 Broker 在线"
        ((unhealthy_count++))
    fi
    
    # 检查是否有下线分区
    offline_partitions=$($KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --describe | grep -c "Leader: none")
    if [ "$offline_partitions" -gt 0 ]; then
        log "WARNING: 存在 $offline_partitions 个下线分区"
        ((unhealthy_count++))
    fi
    
    # 检查未同步副本
    under_replicated=$($KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --describe | grep -c "UnderReplicated")
    if [ "$under_replicated" -gt 0 ]; then
        log "WARNING: 存在 $under_replicated 个未同步副本"
        ((unhealthy_count++))
    fi
    
    return $unhealthy_count
}

# 尝试重启故障节点
recover_node() {
    local node_ip=$1
    local node_name=$2
    
    log "尝试恢复 $node_name ($node_ip)"
    
    # 检查节点是否可访问
    if ! ssh -o ConnectTimeout=10 $node_ip "echo test" >/dev/null 2>&1; then
        log "ERROR: 无法连接到 $node_name"
        return 1
    fi
    
    # 重启 Kafka 服务
    if ssh $node_ip "sudo systemctl restart kafka"; then
        log "INFO: 已重启 $node_name 上的 Kafka 服务"
        sleep 30
        
        # 验证服务是否正常启动
        if ssh $node_ip "systemctl is-active kafka" >/dev/null 2>&1; then
            log "SUCCESS: $node_name 恢复成功"
            return 0
        else
            log "ERROR: $node_name 重启后服务仍未正常"
            return 1
        fi
    else
        log "ERROR: 重启 $node_name 失败"
        return 1
    fi
}

# 主恢复逻辑
main() {
    log "开始集群健康检查"
    
    if check_cluster_health; then
        log "INFO: 集群状态正常"
        exit 0
    else
        log "WARNING: 检测到集群异常，开始自动恢复流程"
    fi
    
    # 检查并恢复各节点
    NODES=("192.168.192.211" "192.168.192.212" "192.168.192.213")
    NODE_NAMES=("ubuntu-kafka-node1" "ubuntu-kafka-node2" "ubuntu-kafka-node3")
    
    for i in "${!NODES[@]}"; do
        node_ip="${NODES[$i]}"
        node_name="${NODE_NAMES[$i]}"
        
        # 检查节点状态
        if ! ssh $node_ip "systemctl is-active kafka" >/dev/null 2>&1; then
            recover_node "$node_ip" "$node_name"
        fi
    done
    
    # 最终状态检查
    sleep 60
    if check_cluster_health; then
        log "SUCCESS: 集群自动恢复完成"
    else
        log "ERROR: 集群自动恢复失败，需要人工干预"
        exit 1
    fi
}

# 执行主函数
main "$@"

使用方法：

chmod +x /opt/kafka/auto-recovery.sh
# 手动执行
/opt/kafka/auto-recovery.sh
# 或配置为定时任务
echo "*/5 * * * * /opt/kafka/auto-recovery.sh" | crontab -

12. 三节点集群最佳实践

12.1 生产环境配置建议

12.1.1 硬件配置标准

组件	每节点配置	集群总计	说明
CPU	4 核+	12 核+	建议 Intel Xeon 或 AMD EPYC
内存	8GB+	24GB+	建议 16GB 以获得更好性能
存储	500GB+ SSD	1.5TB+	NVMe SSD 性能最佳
网络	10Gbps+	30Gbps+	建议万兆网络

12.1.2 集群配置优化

# 三节点集群推荐配置
# 基础配置
default.replication.factor=3
min.insync.replicas=2
num.partitions=6

# 存储配置
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.bytes=107374182400  # 100GB 限制

# 性能配置
num.network.threads=8
num.io.threads=16
background.threads=10

# 刷盘配置
log.flush.interval.messages=10000
log.flush.interval.ms=1000

# 副本配置
replica.lag.time.max.ms=30000
replica.socket.timeout.ms=30000

12.2 运维规范与监控

12.2.1 核心监控指标

Broker 级别指标：

在线 Broker 数量（应始终 = 3）
Controller 活动状态
每秒入站/出站字节数
消息生产和消费速率
请求处理延迟

分区级别指标：

分区数量和 Leader 分布
ISR（In-Sync Replicas）列表状态
Under-replicated 分区数量
Offline 分区数量

系统级别指标：

磁盘使用率（< 80% 正常，> 90% 危险）
JVM 内存使用率和 GC 频率
网络 I/O 和延迟
CPU 使用率

12.2.2 告警规则配置

# Prometheus 告警规则示例
groups:
  - name: kafka-cluster
    rules:
      # 集群健康告警
      - alert: KafkaClusterDown
        expr: count(kafka_server_KafkaServer_BrokerState == 3) < 3
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Kafka 集群节点不足"
          description: "当前只有 {{ $value }} 个节点在线，少于 3 个"

      - alert: KafkaControllerDown
        expr: kafka_controller_KafkaController_ActiveControllerCount < 1
        for: 30s
        labels:
          severity: critical
        annotations:
          summary: "Kafka Controller 不可用"
          description: "集群中没有活跃的 Controller"

      # 数据安全告警
      - alert: KafkaUnderReplicatedPartitions
        expr: kafka_server_ReplicaManager_UnderReplicatedPartitions > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "存在未同步副本"
          description: "有 {{ $value }} 个分区的副本未同步"

      - alert: KafkaOfflinePartitions
        expr: kafka_controller_OfflinePartitionsCount > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "存在下线分区"
          description: "有 {{ $value }} 个分区处于下线状态"

      # 性能告警
      - alert: KafkaHighConsumerLag
        expr: kafka_consumergroup_lag > 10000
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "消费者延迟过高"
          description: "消费者组 {{ $labels.consumergroup }} 延迟超过阈值"

      - alert: KafkaDiskUsageHigh
        expr: (1 - (node_filesystem_avail_bytes{mountpoint="/data/kafka"} / node_filesystem_size_bytes{mountpoint="/data/kafka"})) * 100 > 85
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "磁盘使用率过高"
          description: "Kafka 数据目录磁盘使用率达到 {{ $value }}%"

      - alert: KafkaDiskUsageCritical
        expr: (1 - (node_filesystem_avail_bytes{mountpoint="/data/kafka"} / node_filesystem_size_bytes{mountpoint="/data/kafka"})) * 100 > 95
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "磁盘使用率危险"
          description: "Kafka 数据目录磁盘使用率达到 {{ $value }}%"

12.3 日常运维操作

12.3.1 定期维护任务

每日检查：

# 集群健康检查
/opt/kafka/cluster-health-check.sh

# 检查磁盘使用率
df -h /data/kafka

# 检查 JVM 内存使用
jstat -gc $(pgrep -f kafka.Kafka) 1000 1

每周维护：

# 检查日志清理情况
find /var/log/kafka -name "*.log" -mtime +7 -delete

# 验证数据一致性
/opt/kafka/bin/kafka-topics.sh --bootstrap-server 192.168.192.211:9092 --describe | grep -E "(UnderReplicated|Offline)"

# 备份关键配置
tar -czf /backup/kafka/config_$(date +%Y%m%d).tar.gz /opt/kafka/config

每月维护：

# 完整数据备份
/opt/kafka/backup-kafka.sh

# 性能基准测试
# 使用 kafka-producer-perf-test.sh 和 kafka-consumer-perf-test.sh

# 审计安全配置
# 检查 ACL 和认证配置

12.3.2 应急响应流程

节点故障处理：

确认故障节点（通过健康检查脚本）
检查日志定位根本原因
尝试重启服务
如重启失败，准备替换节点
执行数据恢复（如有必要）
验证集群恢复正常

数据不一致处理：

立即停止相关生产者
检查 ISR 状态
执行分区重新平衡
验证数据完整性
逐步恢复服务

12.4 安全加固建议

12.4.1 网络安全

# 防火墙配置
# 仅允许必要的 IP 访问
iptables -A INPUT -p tcp --dport 9092 -s 192.168.0.0/16 -j ACCEPT
iptables -A INPUT -p tcp --dport 9093 -s 192.168.192.211,192.168.192.212,192.168.192.213 -j ACCEPT
iptables -A INPUT -p tcp --dport 9999 -s 192.168.0.0/16 -j ACCEPT
iptables -A INPUT -j DROP

12.4.2 认证授权

# SASL/SSL 配置示例
listeners=SASL_SSL://:9094
advertised.listeners=SASL_SSL://<YOUR_IP>:9094
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256

# SSL 配置
ssl.keystore.location=/opt/kafka/config/server.keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/opt/kafka/config/server.truststore.jks
ssl.truststore.password=changeit

12.5 性能调优指南

12.5.1 JVM 参数优化

# 针对 8GB 内存节点的 JVM 配置
export KAFKA_HEAP_OPTS="-Xms6g -Xmx6g"
export KAFKA_JVM_OPTS="
  -XX:+UseG1GC
  -XX:MaxGCPauseMillis=20
  -XX:InitiatingHeapOccupancyPercent=35
  -XX:+ExplicitGCInvokesConcurrent
  -XX:MaxInlineLevel=15
  -XX:+UseStringDeduplication
  -Djava.awt.headless=true
"

12.5.2 系统参数优化

# 网络优化
echo 'net.core.somaxconn = 1024' >> /etc/sysctl.conf
echo 'net.core.netdev_max_backlog = 5000' >> /etc/sysctl.conf
echo 'net.ipv4.tcp_max_syn_backlog = 1024' >> /etc/sysctl.conf

# 磁盘 I/O 优化
echo 'deadline' > /sys/block/sdb/queue/scheduler
echo 'vm.dirty_ratio = 15' >> /etc/sysctl.conf
echo 'vm.dirty_background_ratio = 5' >> /etc/sysctl.conf

# 应用配置
sysctl -p

通过遵循这些最佳实践，可以确保 Kafka 三节点集群的稳定性、安全性和高性能运行。

13. 附录

13.1 三节点集群配置参数详解

13.1.1 核心配置参数

参数	三节点推荐值	说明
`node.id`	1,2,3	各节点唯一标识
`process.roles`	broker,controller	每个节点同时担任两种角色
`controller.quorum.voters`	1@ip1:9093,2@ip2:9093,3@ip3:9093	三节点仲裁组配置
`default.replication.factor`	3	确保三副本
`min.insync.replicas`	2	至少两个副本同步
`num.partitions`	6	合理的分区数量

13.1.2 性能优化参数

参数	三节点推荐值	说明
`num.network.threads`	8	网络处理线程数
`num.io.threads`	16	I/O 处理线程数
`log.segment.bytes`	1073741824	1GB 日志段大小
`log.retention.hours`	168	7天数据保留
`log.flush.interval.messages`	10000	消息刷盘间隔

13.2 常用管理命令

13.2.1 集群管理命令

# 集群状态检查
./bin/kafka-metadata-quorum.sh --bootstrap-server 192.168.192.211:9092 describe --status

# 查看所有 Topic
./bin/kafka-topics.sh --bootstrap-server 192.168.192.211:9092 --list

# 创建三副本 Topic
./bin/kafka-topics.sh --create \
  --bootstrap-server 192.168.192.211:9092 \
  --topic my-topic \
  --partitions 6 \
  --replication-factor 3

# 查看 Topic 详情
./bin/kafka-topics.sh --describe \
  --bootstrap-server 192.168.192.211:9092 \
  --topic my-topic

# 检查未同步副本
./bin/kafka-topics.sh --bootstrap-server 192.168.192.211:9092 --describe | grep "UnderReplicated"

13.2.2 故障诊断命令

# 查看 Broker API 版本
./bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.192.211:9092

# 查看日志目录使用情况
./bin/kafka-log-dirs.sh --bootstrap-server 192.168.192.211:9092 --describe

# 查看消费者组状态
./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.192.211:9092 --list

# 重置消费者偏移量
./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.192.211:9092 \
  --group my-group --reset-offsets --to-latest --execute --topic my-topic

13.2.3 性能测试命令

# 生产者性能测试
./bin/kafka-producer-perf-test.sh \
  --topic test-perf \
  --num-records 1000000 \
  --record-size 1000 \
  --throughput -1 \
  --producer-props bootstrap.servers=192.168.192.211:9092

# 消费者性能测试
./bin/kafka-consumer-perf-test.sh \
  --bootstrap-server 192.168.192.211:9092 \
  --topic test-perf \
  --messages 1000000

13.3 监控集成配置

13.3.1 JMX Exporter 配置

# jmx-exporter 配置文件
startDelaySeconds: 0
hostPort: 127.0.0.1:9999
jmxUrl: service:jmx:rmi:///jndi/rmi://127.0.0.1:9999/jmxrmi
ssl: false

rules:
  - pattern: 'kafka.server<type=(.+), name=(.+)PerSec\w*><>Count'
    name: kafka_server_$1_$2_total
    type: COUNTER
    
  - pattern: 'kafka.controller<type=(.+), name=(.+)><>Value'
    name: kafka_controller_$1_$2
    type: GAUGE
    
  - pattern: 'java.lang<type=Memory><>HeapMemoryUsage'
    name: jvm_memory_heap_bytes
    type: GAUGE

13.3.2 Prometheus 抓取配置

scrape_configs:
  - job_name: 'kafka-cluster'
    static_configs:
      - targets:
          - '192.168.192.211:9999'
          - '192.168.192.212:9999'
          - '192.168.192.213:9999'
    scrape_interval: 15s
    scrape_timeout: 10s

13.4 故障排除参考

13.4.1 常见问题及解决方案

问题 1：Controller 无法选举

检查：网络连通性、时间同步、配置一致性
解决：重启所有节点，确保配置正确

问题 2：副本不同步

检查：磁盘空间、网络延迟、节点负载
解决：清理磁盘空间，优化网络配置

问题 3：性能下降

检查：JVM GC、磁盘 I/O、网络瓶颈
解决：调整 JVM 参数，优化系统配置

13.5 资源链接

Apache Kafka 官方文档: https://kafka.apache.org/documentation/
KRaft 模式指南: https://kafka.apache.org/documentation/#kraft
Kafka 最佳实践: https://docs.confluent.io/platform/current/kafka/deployment.html
社区支持: https://stackoverflow.com/questions/tagged/apache-kafka

概述​

为什么选择三节点集群？​

为什么选择 KRaft 模式？​

集群架构优势​

目录​

1. 架构设计​

1.1 三节点高可用架构​

1.2 KRaft 模式组件说明​

1.3 高可用机制​

1.3.1 Raft 仲裁机制​

1.3.2 数据副本机制​

1.3.3 故障恢复机制​

2. 前置条件​

2.1 硬件要求​

2.2 软件要求​

2.3 网络规划​

2.4 端口规划​

3. 手动部署指南​

3.1 环境准备​

3.1.1 系统初始化（所有节点）​

3.1.2 配置时间同步​

3.1.3 配置防火墙​

3.1.4 配置主机名解析​

3.2 安装 Java​

3.3 安装 Kafka​

3.4 配置 Kafka​

3.4.1 节点1配置（ubuntu-kafka-node1）​

3.4.2 节点2配置（ubuntu-kafka-node2）​

3.4.3 节点3配置（ubuntu-kafka-node3）​

3.5 创建系统服务​

3.6 集群初始化​

3.6.1 生成集群 ID​

3.6.2 格式化存储目录​

3.5.3 启动 Kafka 服务​

3.7 集群验证​

3.7.1 检查集群状态​

3.4.2 创建测试主题​

3.4.3 生产/消费消息测试​

3.4.4 高可用性验证​

3.5 集群健康检查​

5. 网络配置​

5.1 端口配置​

5.2 监听器配置​

5.2.1 PLAINTEXT 监听器​

5.2.2 SSL 监听器​

5.2.3 SASL 认证​

5.3 外部访问配置​

5.3.1 跨网络访问​

5.3.2 负载均衡配置​

6.2.1 PLAINTEXT 监听器​

5.2.2 SSL 监听器​

5.2.3 SASL 认证​

5.3 外部访问配置​

5.3.1 跨网络访问​

5.3.2 负载均衡配置​

7. 持久化存储配置​

7.1 存储目录规划​

7.2 日志段配置​

7.3 磁盘 I/O 优化​

7.3.1 文件系统选择​

7.3.2 磁盘调度器​

7.3.3 RAID 配置​

8. 监控与告警​

8.1 JMX 监控配置​

8.1.1 启用 JMX 端口​

8.1.2 验证 JMX 连接​

8.2 Prometheus 监控​

8.2.1 部署 JMX Exporter​

8.2.2 配置 Prometheus 抓取​

8.2.3 Grafana 仪表盘​

8.3 告警规则​

9. 运维操作​

9.1 日常运维​

9.1.1 Topic 管理​

9.1.2 消费组管理​

9.1.3 日志清理​

9.2 扩缩容​

9.2.1 增加 Broker 节点​

9.2.2 减少 Broker 节点​

9.2.3 重平衡分区​

概述

为什么选择三节点集群？

为什么选择 KRaft 模式？

集群架构优势

目录

1. 架构设计

1.1 三节点高可用架构

1.2 KRaft 模式组件说明

1.3 高可用机制

1.3.1 Raft 仲裁机制

1.3.2 数据副本机制

1.3.3 故障恢复机制

2. 前置条件

2.1 硬件要求

2.2 软件要求

2.3 网络规划

2.4 端口规划

3. 手动部署指南

3.1 环境准备

3.1.1 系统初始化（所有节点）

3.1.2 配置时间同步

3.1.3 配置防火墙

3.1.4 配置主机名解析

3.2 安装 Java

3.3 安装 Kafka

3.4 配置 Kafka

3.4.1 节点1配置（ubuntu-kafka-node1）

3.4.2 节点2配置（ubuntu-kafka-node2）

3.4.3 节点3配置（ubuntu-kafka-node3）

3.5 创建系统服务

3.6 集群初始化

3.6.1 生成集群 ID

3.6.2 格式化存储目录

3.5.3 启动 Kafka 服务

3.7 集群验证

3.7.1 检查集群状态

3.4.2 创建测试主题

3.4.3 生产/消费消息测试

3.4.4 高可用性验证

3.5 集群健康检查

5. 网络配置

5.1 端口配置

5.2 监听器配置

5.2.1 PLAINTEXT 监听器

5.2.2 SSL 监听器

5.2.3 SASL 认证

5.3 外部访问配置

5.3.1 跨网络访问

5.3.2 负载均衡配置

6.2.1 PLAINTEXT 监听器

5.2.2 SSL 监听器

5.2.3 SASL 认证

5.3 外部访问配置

5.3.1 跨网络访问

5.3.2 负载均衡配置

7. 持久化存储配置

7.1 存储目录规划

7.2 日志段配置

7.3 磁盘 I/O 优化

7.3.1 文件系统选择

7.3.2 磁盘调度器

7.3.3 RAID 配置

8. 监控与告警

8.1 JMX 监控配置

8.1.1 启用 JMX 端口

8.1.2 验证 JMX 连接

8.2 Prometheus 监控

8.2.1 部署 JMX Exporter

8.2.2 配置 Prometheus 抓取

8.2.3 Grafana 仪表盘

8.3 告警规则

9. 运维操作

9.1 日常运维

9.1.1 Topic 管理

9.1.2 消费组管理

9.1.3 日志清理

9.2 扩缩容

9.2.1 增加 Broker 节点

9.2.2 减少 Broker 节点

9.2.3 重平衡分区