Kafka 集群部署

系统配置调整

关闭防火墙

sudo systemctl stop ufw
sudo systemctl disable ufw

资源配置调整

编辑/etc/security/limits.conf文件

sudo vi /etc/security/limits.conf

添加以下内容：

* soft nofile 65536
* hard nofile 65536
* soft nproc 65536
* hard nproc 65536

安装与配置 Kafka 集群

下载并解压

cd /export/software
wget -P /export/software/ https://repo.huaweicloud.com/apache/kafka/3.4.0/kafka_2.12-3.4.0.tgz
tar -zxvf /export/software/kafka_2.12-3.4.0.tgz -C /export/servers/
cd /export/servers
mv /export/servers/kafka_2.12-3.4.0 /export/servers/kafka

配置 Kafka 服务器参数

编辑修改/export/servers/kafka/config/server.properties文件

sudo chown -R hadooper:hadooper /export/servers/kafka/
vi /export/servers/kafka/config/server.properties

hadoop01：

broker.id=1
listeners=PLAINTEXT://hadoop01:9092
advertised.listeners=PLAINTEXT://hadoop01:9092
log.dirs=/export/servers/kafka/kafka-logs
num.partitions=3
zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
default.replication.factor=3
offsets.topic.replication.factor=3

vi /export/servers/kafka/bin/kafka-server-start.sh

添加如下内容于脚本开始处：

export JAVA_HOME=/export/servers/jdk

vi /export/servers/kafka/bin/kafka-server-stop.sh

添加如下内容于脚本开始处：

export JAVA_HOME=/export/servers/jdk

删除并替换 hive 中原有的 kafka：

rm /export/servers/hive-4.0.1/lib/kafka-clients-2.5.0.jar
cp /export/servers/kafka/libs/kafka-clients-3.4.0.jar /export/servers/hive-4.0.1/lib/

于 hadoop01 节点分发至 hadoop02、hadoop03 节点

scp -r /export/servers/kafka hadoop02:/tmp/
scp -r /export/servers/kafka hadoop03:/tmp/

进入 hadoop02:

sudo cp -r /tmp/kafka /export/servers/
rm -r /tmp/kafka

编辑修改/export/servers/kafka/config/server.properties文件

sudo chown -R hadooper:hadooper /export/servers/kafka/
vi /export/servers/kafka/config/server.properties

hadoop02:

broker.id=2
listeners=PLAINTEXT://hadoop02:9092
advertised.listeners=PLAINTEXT://hadoop02:9092
log.dirs=/export/servers/kafka/kafka-logs
num.partitions=3
zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
default.replication.factor=3
offsets.topic.replication.factor=3

进入 hadoop03:

sudo cp -r /tmp/kafka /export/servers/
rm -r /tmp/kafka

编辑修改/export/servers/kafka/config/server.properties文件

sudo chown -R hadooper:hadooper /export/servers/kafka/
vi /export/servers/kafka/config/server.properties

hadoop03:

broker.id=3
listeners=PLAINTEXT://hadoop03:9092
advertised.listeners=PLAINTEXT://hadoop03:9092
log.dirs=/export/servers/kafka/kafka-logs
num.partitions=3
zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181
default.replication.factor=3
offsets.topic.replication.factor=3

启动集群验证

启动 ZooKeeper 集群：在各节点上启动 ZooKeeper 服务

启动节点

zkServer.sh start

检查节点角色（Leader/Follower）

zkServer.sh status

启动 Kafka 集群

按顺序在每个 Kafka 节点上，使用以下命令启动 Kafka 服务：

cd /export/servers/kafka
mkdir -p /export/servers/kafka/logs
mkdir -p /export/servers/kafka/kafka-logs
/export/servers/kafka/bin/kafka-server-start.sh -daemon /export/servers/kafka/config/server.properties

测试

jps

出现同时出现 QuorumPeerMain 和 Kafka 则代表服务启动成功,如下图所示

测试集群工作状态

在 hadoop01 上创建 topic:

/export/servers/kafka/bin/kafka-topics.sh --create --bootstrap-server hadoop01:9092,hadoop02:9092,hadoop03:9092 --replication-factor 3 --partitions 3 --topic my-test-topic

查看 topic 描述确认分区和副本分配情况

/export/servers/kafka/bin/kafka-topics.sh --describe --bootstrap-server hadoop01:9092 --topic my-test-topic

验证生产消费端是否消息互通

hadoop01(生产者):

/export/servers/kafka/bin/kafka-console-producer.sh --bootstrap-server hadoop01:9092 --topic my-test-topic

hadoop02(消费者):

/export/servers/kafka/bin/kafka-console-consumer.sh --bootstrap-server hadoop02:9092 --topic my-test-topic --from-beginning

hadoo03(消费者):

/export/servers/kafka/bin/kafka-console-consumer.sh --bootstrap-server hadoop03:9092 --topic my-test-topic --from-beginning

测试集群状态

在 hadoop01 输入数据，在 hadoop02、hadoop03 查看数据，若全部互通则代表 kafka 集群正常运行，如下图所示