kafka集群部署

1 在三台 CentOS 机器上搭建 Kafka 集群（传统模式）

在三台 CentOS 机器（192.168.1.102、192.168.1.103、192.168.1.104）上搭建 Kafka 3.0.0 集群的详细步骤。

1.1 环境准备

确保三台机器满足以下条件：

• CentOS 系统
• JDK 1.8 已安装
• 网络互通，主机名解析正常
• 防火墙已关闭或相关端口已开放

1.2 在所有机器上安装 Kafka

1.2.1. 下载并解压 Kafka

在所有三台机器上执行以下命令：

# 下载 Kafka
wget https://archive.apache.org/dist/kafka/3.0.0/kafka_2.12-3.0.0.tgz# 解压到 /usr/local 目录
tar -xzf kafka_2.12-3.0.0.tgz -C /usr/local/
cd /usr/local/kafka_2.12-3.0.0

1.2.2. 创建数据目录

在所有机器上创建 Kafka 数据目录：

mkdir -p /data/kafka-logs
mkdir -p /data/zookeeper

1.3 配置 ZooKeeper 集群

Kafka 使用 ZooKeeper 来管理集群元数据。我们将配置一个三节点的 ZooKeeper 集群。

1.3.1. 配置 ZooKeeper

在所有三台机器上编辑 ZooKeeper 配置文件：

vim config/zookeeper.properties

添加以下内容（注意每台机器的 server.X 配置要相同）：

dataDir=/data/zookeeper
clientPort=2181
maxClientCnxns=0
admin.enableServer=false# ZooKeeper 集群配置
server.1=192.168.1.102:2888:3888
server.2=192.168.1.103:2888:3888
server.3=192.168.1.104:2888:3888# 其他优化配置
tickTime=2000
initLimit=10
syncLimit=5

1.3.2. 创建 myid 文件

在每个节点的 /data/zookeeper 目录下创建 myid 文件，内容分别为 1、2、3：

在 192.168.1.102 上：

echo 1 > /data/zookeeper/myid

在 192.168.1.103 上：

echo  2  > /data/zookeeper/myid

在 192.168.1.104 上：

echo  3  > /data/zookeeper/myid

1.4 配置 Kafka 集群

1.4.1. 配置 Kafka Broker

在每个节点上编辑 Kafka 配置文件：

vim config/server.properties

在 192.168.1.102 上配置：

broker.id=1
listeners=PLAINTEXT://192.168.1.102:9092
advertised.listeners=PLAINTEXT://192.168.1.102:9092
log.dirs=/data/kafka-logs
zookeeper.connect=192.168.1.102:2181,192.168.1.103:2181,192.168.1.104:2181
num.partitions=3
default.replication.factor=3
min.insync.replicas=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

在 192.168.1.103 上配置：

broker.id=2
listeners=PLAINTEXT://192.168.1.103:9092
advertised.listeners=PLAINTEXT://192.168.1.103:9092
log.dirs=/data/kafka-logs
zookeeper.connect=192.168.1.102:2181,192.168.1.103:2181,192.168.1.104:2181
num.partitions=3
default.replication.factor=3
min.insync.replicas=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

在 192.168.1.104 上配置：

broker.id=3
listeners=PLAINTEXT://192.168.1.104:9092
advertised.listeners=PLAINTEXT://192.168.1.104:9092
log.dirs=/data/kafka-logs
zookeeper.connect=192.168.1.102:2181,192.168.1.103:2181,192.168.1.104:2181
num.partitions=3
default.replication.factor=3
min.insync.replicas=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

1.5 启动集群

1. 启动 ZooKeeper 集群

按照顺序在三台机器上启动 ZooKeeper：

# 在每台机器上执行
nohup bin/zookeeper-server-start.sh config/zookeeper.properties > zookeeper.log 2>&1 &

2. 验证 ZooKeeper 集群状态

在任意一台机器上执行：

echo stat | nc localhost 2181

应该能看到类似 “Mode: leader” 或 “Mode: follower” 的输出。

3. 启动 Kafka 集群

在三台机器上依次启动 Kafka Broker：

# 在每台机器上执行
nohup bin/kafka-server-start.sh config/server.properties > kafka.log 2>&1 &

1.6 验证集群状态

1.6.1. 检查 Broker 是否已加入集群

在任意一台机器上执行：

bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.1.102:9092,192.168.1.103:9092,192.168.1.104:9092

1.6.2. 创建测试 Topic

在任意一台机器上执行：

bin/kafka-topics.sh --create \--bootstrap-server 192.168.1.102:9092 \--replication-factor 3 \--partitions 3 \--topic test-cluster-topic

1.6.3. 查看 Topic 详情

bin/kafka-topics.sh --describe \--bootstrap-server 192.168.1.102:9092 \--topic test-cluster-topic

输出应该显示分区均匀分布在三个 Broker 上，每个分区都有三个副本。

1.6.4. 测试消息生产和消费

在生产节点（例如 192.168.1.102）上启动生产者：

bin/kafka-console-producer.sh \--bootstrap-server 192.168.1.102:9092 \--topic test-cluster-topic

在消费节点（例如 192.168.1.103）上启动消费者：

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

1.7 集群管理脚本

1.7.1. 创建启动脚本

在每台机器上创建启动脚本 start-cluster.sh：

#!/bin/bash
echo "Starting ZooKeeper..."
nohup /usr/local/kafka_2.12-3.0.0/bin/zookeeper-server-start.sh /usr/local/kafka_2.12-3.0.0/config/zookeeper.properties > /usr/local/kafka_2.12-3.0.0/zookeeper.log 2>&1 &sleep 5echo "Starting Kafka..."
nohup /usr/local/kafka_2.12-3.0.0/bin/kafka-server-start.sh /usr/local/kafka_2.12-3.0.0/config/server.properties > /usr/local/kafka_2.12-3.0.0/kafka.log 2>&1 &echo "Services started. Check logs for details."

1.7.2. 创建停止脚本

在每台机器上创建停止脚本 stop-cluster.sh：

#!/bin/bash
echo "Stopping Kafka..."
/usr/local/kafka_2.12-3.0.0/bin/kafka-server-stop.shsleep 5echo "Stopping ZooKeeper..."
/usr/local/kafka_2.12-3.0.0/bin/zookeeper-server-stop.sh# 如果上述命令无效，强制停止
KAFKA_PID=$(ps aux | grep -i 'kafka\.Kafka' | grep -v grep | awk '{print $2}')
ZK_PID=$(ps aux | grep -i QuorumPeerMain | grep -v grep | awk '{print $2}')if [ -n "$KAFKA_PID" ]; thenkill -9 $KAFKA_PID
fiif [ -n "$ZK_PID" ]; thenkill -9 $ZK_PID
fiecho "All services stopped."

1.7.3. 设置脚本权限

chmod +x start-cluster.sh stop-cluster.sh

1.8 防火墙配置

如果系统防火墙开启，需要开放相关端口：

# 开放 ZooKeeper 端口
firewall-cmd --permanent --add-port=2181/tcp
firewall-cmd --permanent --add-port=2888/tcp
firewall-cmd --permanent --add-port=3888/tcp# 开放 Kafka 端口
firewall-cmd --permanent --add-port=9092/tcp# 重新加载防火墙
firewall-cmd --reload

1.9 监控和管理

1. 安装 Kafka 管理工具（可选）

安装 Kafka Manager 或其他监控工具来更方便地管理集群。

2. 配置日志轮转

编辑 /etc/logrotate.d/kafka 文件，添加以下内容：

/usr/local/kafka_2.12-3.0.0/kafka.log {dailyrotate 7compressdelaycompressmissingoknotifemptycreate 644 root root
}

注意事项

1. 启动顺序：先启动 ZooKeeper 集群，再启动 Kafka 集群
2. 停止顺序：先停止 Kafka 集群，再停止 ZooKeeper 集群
3. 数据备份：定期备份 ZooKeeper 和 Kafka 的重要数据
4. 监控：设置监控告警，关注磁盘空间、CPU 和内存使用情况
5. 生产环境：对于生产环境，建议将 ZooKeeper 和 Kafka 分开部署在不同的机器上

2 在三台 CentOS 机器上搭建 Kafka 集群（KRAFT）

前言

在 CentOS 上部署 KRaft 模式的 Kafka 集群确实比传统的 ZooKeeper 依赖模式更简洁，它能提升集群性能和管理效率。下面我将为你提供详细的步骤。

首先，我用一个表格来汇总三台服务器（IP：192.168.1.102, 192.168.1.103, 192.168.1.104）的关键配置，以便你有一个全局视图：

配置项	192.168.1.102	192.168.1.103	192.168.1.104
node.id	1	2	3
process.roles	broker,controller	broker,controller	broker,controller
controller.quorum.voters	1@192.168.1.102:9093,2@192.168.1.103:9093,3@192.168.1.104:9093 (所有节点相同)	同左	同左
listeners	PLAINTEXT://:9092,CONTROLLER://:9093	PLAINTEXT://:9092,CONTROLLER://:9093	PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners	PLAINTEXT://192.168.1.102:9092	PLAINTEXT://192.168.1.103:9092	PLAINTEXT://192.168.1.104:9092
log.dirs	/data/kafka-logs	/data/kafka-logs	/data/kafka-logs

接下来是详细的部署步骤。

2.1 环境准备

在所有三台服务器（192.168.1.102, 192.168.1.103, 192.168.1.104）上操作：

1. 安装 JDK 1.8：Kafka 需要 Java 运行环境。

   yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
   

   验证安装：

   java -version
   

2. 设置主机名解析（可选但建议）：确保每台服务器都能解析其他服务器的主机名。编辑 /etc/hosts 文件，添加：

   192.168.1.102 kafka-node1
   192.168.1.103 kafka-node2
   192.168.1.104 kafka-node3
   

3. 防火墙配置：开放 Kafka 和 Controller 所需的端口。

   firewall-cmd --permanent --add-port=9092/tcp  # Kafka Broker 端口
   firewall-cmd --permanent --add-port=9093/tcp  # Controller 端口
   firewall-cmd --reload
   

2.2 安装 Kafka

在所有三台服务器上下载并解压 Kafka：

1. 下载 Kafka（以 3.0.0 版本为例）：

   wget https://archive.apache.org/dist/kafka/3.0.0/kafka_2.12-3.0.0.tgz
   

2. 解压到指定目录：

   tar -xzf kafka_2.12-3.0.0.tgz -C /usr/local/
   cd /usr/local/kafka_2.12-3.0.0
   

3. 创建数据目录：

   mkdir -p /data/kafka-logs  # 用于存储 Kafka 数据
   

2.3 配置 Kafka KRaft 模式

KRaft 模式是 Kafka 的新架构，它不再依赖 ZooKeeper，而是使用 Kafka 自身的 Raft 协议来管理元数据。

在所有三台服务器上配置 config/kraft/server.properties 文件。请注意，每个节点的 node.id 和 advertised.listeners 需要不同。

在 192.168.1.102 上配置：

# 节点唯一标识
node.id=1
# 节点角色（同时作为broker和controller）
process.roles=broker,controller
# 监听器配置
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
# 对外公告的监听地址（客户端使用）
advertised.listeners=PLAINTEXT://192.168.1.102:9092
# 控制器仲裁投票者列表（所有节点相同）
controller.quorum.voters=1@192.168.1.102:9093,2@192.168.1.103:9093,3@192.168.1.104:9093
# 数据日志目录
log.dirs=/data/kafka-logs
# 监听器安全协议映射
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
# 控制器监听器名称
controller.listener.names=CONTROLLER
# 内部Broker通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT

在 192.168.1.103 上配置：

node.id=2
process.roles=broker,controller
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://192.168.1.103:9092
controller.quorum.voters=1@192.168.1.102:9093,2@192.168.1.103:9093,3@192.168.1.104:9093
log.dirs=/data/kafka-logs
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
controller.listener.names=CONTROLLER
inter.broker.listener.name=PLAINTEXT

在 192.168.1.104 上配置：

node.id=3
process.roles=broker,controller
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://192.168.1.104:9092
controller.quorum.voters=1@192.168.1.102:9093,2@192.168.1.103:9093,3@192.168.1.104:9093
log.dirs=/data/kafka-logs
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
controller.listener.names=CONTROLLER
inter.broker.listener.name=PLAINTEXT

2.4 初始化并启动集群

1. 生成集群 UUID（在任意一台服务器上执行一次即可）：

   /usr/local/kafka_2.12-3.0.0/bin/kafka-storage.sh random-uuid
   

   记下输出的 UUID（例如 4hU1sxPUQG-4qqqgZQxFZA），下一步会用到。

2. 格式化存储目录（在所有三台服务器上执行，使用相同的 UUID）：

   /usr/local/kafka_2.12-3.0.0/bin/kafka-storage.sh format -t <生成的UUID> -c /usr/local/kafka_2.12-3.0.0/config/kraft/server.properties
   

   成功后会显示 “Formatting /data/kafka-logs”。

3. 启动 Kafka 服务（在所有三台服务器上执行）：

   cd /usr/local/kafka_2.12-3.0.0
   nohup bin/kafka-server-start.sh config/kraft/server.properties > kafka.log 2>&1 &
   

2.5 验证集群状态

1. 检查进程：

   jps -l | grep kafka
   

   应该能看到 kafka.Kafka 进程。

2. 检查 Broker 是否加入集群（在任意一台服务器上执行）：

   bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.1.102:9092
   

3. 查看元数据仲裁状态（在任意一台服务器上执行）：

   bin/kafka-metadata-quorum.sh describe --bootstrap-server 192.168.1.102:9092
   

   输出中应显示一个 Leader 和两个 Follower，并且所有节点状态为 Online。

4. 创建测试 Topic：

   bin/kafka-topics.sh --create \
   --bootstrap-server 192.168.1.102:9092 \
   --replication-factor 3 \
   --partitions 3 \
   --topic test-kraft-topic
   

5. 查看 Topic 详情：

   bin/kafka-topics.sh --describe \
   --bootstrap-server 192.168.1.102:9092 \
   --topic test-kraft-topic
   

   输出应显示分区和副本均匀分布在三个 Broker 上。

2.6 生产环境建议

• 角色分离：对于生产环境，考虑将 Controller 角色和 Broker 角色分离部署，以避免资源竞争。你可以让一些节点专门运行 Controller 角色 (process.roles=controller)，另一些节点专门运行 Broker 角色 (process.roles=broker)。
• 控制器节点数量：Controller 节点数量建议为 3 或 5 个（奇数），以便容忍 (n-1)/2 个节点故障。
• 资源分配：确保为 Kafka 进程分配足够的堆内存。可以通过设置环境变量 KAFKA_HEAP_OPTS 来实现，例如：

  export KAFKA_HEAP_OPTS="-Xmx4G -Xms4G"
  

• 日志保留策略：根据磁盘空间和需求配置日志保留策略，例如在 server.properties 中设置：

  log.retention.hours=168
  

• 监控：关注 Controller 选举耗时、元数据复制延迟、Broker 磁盘 I/O 等关键指标。

2.7 停止集群

当需要停止集群时，在每个节点上执行：

/usr/local/kafka_2.12-3.0.0/bin/kafka-server-stop.sh

2.8 注意事项

• 集群 ID：确保在格式化存储目录时使用相同的集群 ID。
• 防火墙：确保所有服务器之间的 9092 和 9093 端口是互通的。
• 时间同步：确保所有服务器的时间同步，使用 NTP 服务。
• 磁盘空间：监控 /data/kafka-logs 目录的磁盘空间使用情况。
• 配置文件：仔细检查每个节点的配置文件，确保 node.id 和 advertised.listeners 的正确性。

通过以上步骤，你应该能够在三台 CentOS 服务器上成功搭建一个 KRaft 模式的 Kafka 3.0.0 集群。KRaft 模式简化了 Kafka 的架构，不再需要单独部署和维护 ZooKeeper 集群，降低了运维复杂度。