1、介绍
本集群采用三节点模式，jdk版本为openjdk-8
kafka虽然从2.8开始可以去zookeeper用Kafka Raft元数据模式启动，但是如果用于生产环境，还是老老实实用zookeeper，避免出现一些棘手的问题。
涉及zookeeper集群这一块的内容您可以选择和本教程一样选择kafka自带的zookeeper，也可以用本人其他教程来替换zookeeper集群安装
ip与hostname对应关系
kafka-1 对应 192.168.0.201
kafka-2 对应 192.168.0.202
kafka-3 对应 192.168.0.203
2、修改hostname
在节点1执行
hostnamectl set-hostname kafka-1
在节点2执行
hostnamectl set-hostname kafka-2
在节点3执行
hostnamectl set-hostname kafka-3
3、分别在三台机器上执行如下操作关闭防火墙、停止防火墙并禁止开启启动防火墙：
systemctl stop firewalld.service&&systemctl disable firewalld.service
关闭selinux ：
vi /etc/selinux/config
将SELINUX=enforcing改为SELINUX=disabled
在三台机器上执行reboot重启三台机器
reboot
4、重启完成之后重新连接机器，在三台机器上执行下面的命令安装openjdk-8(自行选择下面的命令安装jdk8或者jdk11)和wget
jdk8
dnf -y install wget java-1.8.0-openjdk java-1.8.0-openjdk-devel
jdk11
dnf -y install wget java-11-openjdk java-11-openjdk-devel
验证jdk和wget是否安装成功
在三台机器上分别执行以下命令，如果显示和下图一样，说明jdk和wget已安装成功
java -version&&wget
5、修改三台机器的hosts
vi /etc/hosts
添加以下内容到/etc/hosts文件里面
192.168.0.201 kafka-1
192.168.0.202 kafka-2
192.168.0.203 kafka-3
测试host是否生效，在3台机器上执行以下命令：
ping kafka-1
ping kafka-2
ping kafka-3
显示如图片，说明host配置成功
6、下载kafka,以下命令在第一节点执行即可
mkdir /tmp/kafka
cd /tmp/kafka
wget https://dlcdn.apache.org/kafka/3.1.0/kafka_2.12-3.1.0.tgz
tar -zxf kafka_2.12-3.1.0.tgz
ls
7、移动文件夹至opt目录下,以下命令在第一节点执行即可
mv /tmp/kafka/kafka_2.12-3.1.0/ /opt/
ls /opt/
8、建立zookeeper的data目录,以下命令在第一节点执行即可
mkdir -p /opt/kafka_2.12-3.1.0/data/zookeeper
9、修改zookeeper配置文件,以下命令在第一节点执行即可
vi /opt/kafka_2.12-3.1.0/config/zookeeper.properties
修改dataDir为/opt/kafka_2.12-3.1.0/data/zookeeper，添加以下三行到文件末尾
dataDir=/opt/kafka_2.12-3.1.0/data/zookeeper
server.1=kafka-1:2888:3888
server.2=kafka-2:2888:3888
server.3=kafka-3:2888:3888
tickTime=2000
initLimit=10
syncLimit=5
maxClientCnxns=120
10、建立kafka-logs目录,以下命令在第一节点执行即可
mkdir -p /opt/kafka_2.12-3.1.0/data/kafka-logs
11、修改kafka配置文件
vi /opt/kafka_2.12-3.1.0/config/server.properties
修改broker.id为1,以下命令在第一节点执行即可
broker.id=1
修改log.dirs目录为/opt/kafka_2.12-3.1.0/data/kafka-logs,以下命令在第一节点执行即可
log.dirs=/opt/kafka_2.12-3.1.0/data/kafka-logs
修改num.partitions为3,以下命令在第一节点执行即可
num.partitions=3
修改zookeeper.connect,以下命令在第一节点执行即可
zookeeper.connect=kafka-1:2181,kafka-2:2181,kafka-3:2181
12、复制安装文件至kafka-2、kafka-3
注意：下面的命令需要输入ssh密码
scp -r /opt/kafka_2.12-3.1.0/ kafka-2:/opt/
scp -r /opt/kafka_2.12-3.1.0/ kafka-3:/opt/
13、写入zookeeper的myid
在kafka-1上执行
echo "1" > /opt/kafka_2.12-3.1.0/data/zookeeper/myid
在kafka-2上执行
echo "2" > /opt/kafka_2.12-3.1.0/data/zookeeper/myid
在kafka-3上执行
echo "3" > /opt/kafka_2.12-3.1.0/data/zookeeper/myid
14、修改kafka-2的broker.id为2
vi /opt/kafka_2.12-3.1.0/config/server.properties
broker.id=2
15、修改kafka-3的broker.id为3
vi /opt/kafka_2.12-3.1.0/config/server.properties
broker.id=3
16、在三个节点上分别执行以下命令启动zookeeper，执行过程中按回车退出控制台输出
/opt/kafka_2.12-3.1.0/bin/zookeeper-server-start.sh /opt/kafka_2.12-3.1.0/config/zookeeper.properties &
17、在三台上执行jps命令，如果有QuorumPeerMain，说明zookeeper启动成功
jps
18、在三个节点上分别执行以下命令启动kafka，执行过程中按回车退出控制台输出
/opt/kafka_2.12-3.1.0/bin/kafka-server-start.sh -daemon /opt/kafka_2.12-3.1.0/config/server.properties &
执行jps查看kafka进程
jps
在节点一执行创建topic，测试kafka可用性
/opt/kafka_2.12-3.1.0/bin/kafka-topics.sh --create --bootstrap-server kafka-1:9092,kafka-2:9092,kafka-3:9092 --replication-factor 3 --partitions 3 --topic test
在节点一执行删除topic
/opt/kafka_2.12-3.1.0/bin/kafka-topics.sh --delete --bootstrap-server kafka-1:9092,kafka-2:9092,kafka-3:9092 --topic test
