CentOS系统下如何从零开始搭建一个完整的Kafka集群？

在CentOS上部署一个高可用的Kafka集群是构建大规模数据管道和实时流处理应用的关键步骤，本文将提供一个清晰、分步的指南,帮助您在CentOS系统上完成Kafka集群的安装与配置。

准备工作

在开始安装之前，请确保您的环境满足以下基本要求,一个典型的Kafka集群至少需要三个节点来保证Zookeeper和Kafka自身的容错性。

服务器环境：
准备至少三台安装了CentOS 7或CentOS 8的服务器，确保它们之间网络互通,并已配置好静态IP地址。

Java环境：
Kafka是基于Java开发的，因此需要在每台服务器上安装Java运行环境（JRE）或Java开发工具包（JDK），建议使用Java 8或更高版本,可以通过以下命令安装OpenJDK：

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

安装完成后，使用 java -version 命令验证是否成功。

配置主机名与hosts解析：
为了方便节点间的识别，为每台服务器设置唯一的主机名，并在所有服务器的 /etc/hosts 文件中添加集群所有节点的解析记录。

# 在每台服务器上执行，
sudo hostnamectl set-hostname kafka1
# 编辑 /etc/hosts 文件，添加如下内容：
192.168.1.101 kafka1
192.168.1.102 kafka2
192.168.1.103 kafka3

防火墙配置：
Kafka和Zookeeper需要开放多个端口，请根据您的安全策略,在防火墙中放行以下端口：

• Zookeeper: 2181 (客户端连接), 2888 (集群内部通信), 3888 (集群选举)
• Kafka: 9092 (Broker与客户端通信)

使用 firewall-cmd 命令开放端口：

sudo firewall-cmd --permanent --add-port={2181,2888,3888,9092}/tcp
sudo firewall-cmd --reload

安装与配置

下载并解压Kafka：
在所有节点上，从官方网站下载稳定版的Kafka并解压，这里以2.8.1版本为例。

wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.13-2.8.1.tgz
tar -xzf kafka_2.13-2.8.1.tgz
sudo mv kafka_2.13-2.8.1 /usr/local/kafka

配置Zookeeper集群：
Kafka依赖Zookeeper来管理集群元数据,我们需要在每台服务器上配置Zookeeper。

为Zookeeper创建数据目录：

sudo mkdir -p /data/zookeeper

编辑 /usr/local/kafka/config/zookeeper.properties 文件,修改以下内容：

dataDir=/data/zookeeper
# 服务器列表，格式为 server.X=hostname:leader_port:election_port
server.1=kafka1:2888:3888
server.2=kafka2:2888:3888
server.3=kafka3:2888:3888

在每台服务器的 dataDir 目录下创建一个名为 myid 的文件，文件内容即为该节点在 server.X 中对应的X值。

# 在 kafka1 上
echo "1" | sudo tee /data/zookeeper/myid
# 在 kafka2 上
echo "2" | sudo tee /data/zookeeper/myid
# 在 kafka3 上
echo "3" | sudo tee /data/zookeeper/myid

配置Kafka Broker：
这是配置Kafka集群的核心，编辑 /usr/local/kafka/config/server.properties 文件,每个节点的配置都需要进行个性化修改。

关键配置项如下：
| 属性 | 描述 | 示例 |
|—|—|—|
| broker.id | 每个Broker在集群中的唯一标识符 | broker.id=1 (kafka1), broker.id=2 (kafka2) |
| listeners | Broker监听的地址和端口 | listeners=PLAINTEXT://kafka1:9092 |
| zookeeper.connect | Zookeeper集群的连接地址 | zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181 |
| log.dirs | Kafka消息日志的存储目录 | log.dirs=/data/kafka-logs |

请确保在每个节点上，broker.id 和 listeners 中的主机名都与当前机器匹配,创建日志目录：

sudo mkdir -p /data/kafka-logs

启动服务并验证：
首先启动所有节点上的Zookeeper,然后启动Kafka。

# 启动Zookeeper
/usr/local/kafka/bin/zookeeper-server-start.sh -daemon /usr/local/kafka/config/zookeeper.properties
# 启动Kafka
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

为了方便管理,建议为Zookeeper和Kafka创建systemd服务单元文件。

验证集群是否正常工作,在任意一个节点上创建一个测试主题：

/usr/local/kafka/bin/kafka-topics.sh --create --topic test-topic --bootstrap-server kafka1:9092 --replication-factor 3 --partitions 3

在一个节点启动生产者，在另一个节点启动消费者，检查消息是否能正常收发,以此验证集群功能。

相关问答FAQs

Q1: 启动Kafka后，客户端无法连接到Broker，可能是什么原因？
A1: 这通常由以下几个问题导致：

1. 防火墙问题：请确保Kafka的端口（默认9092）已在所有节点的防火墙中开放。
2. listeners配置错误：检查 server.properties 中的 listeners 配置，如果配置为 localhost 或 0.0.1，外部客户端将无法访问,应配置为客户端可解析的主机名或IP地址。
3. 网络不通：确认客户端机器与Kafka Broker服务器之间的网络是连通的，可以使用 ping 或 telnet 命令进行排查。
4. Zookeeper连接问题：如果Kafka无法连接到Zookeeper集群，它可能无法正常启动，检查 zookeeper.connect 配置和Zookeeper服务状态。

Q2: 如何向一个正在运行的Kafka集群中添加新的Broker节点？
A2: 添加新Broker是一个在线操作,可以平滑进行：

1. 配置新节点：在新服务器上完成与现有节点相同的准备工作（Java、hosts解析、防火墙）。
2. 安装并配置Kafka：安装Kafka，并为其分配一个全新的、未被使用的 broker.id，配置 listeners 和 zookeeper.connect（指向现有集群）。
3. 启动新Broker：启动新节点的Kafka服务，它会自动注册到Zookeeper，并加入集群,它不会承接任何数据流量。
4. 迁移分区：使用 kafka-reassign-partitions.sh 工具创建一个重分配方案，将部分主题的分区副本迁移到新的Broker上，执行该方案，Kafka会自动在后台进行数据复制，将流量逐步切换到新节点，这个过程是可控的,可以避免对业务造成大的影响。