这里使用kafka自带的zookeeper进行集群搭建，jdk版本要求1.8以上
kafka_2.13-3.0.0.tgz
https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
设置jdk环境变量
vim /etc/profile export JAVA_HOME=/home/jdk-11 export CLASSPATH=$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar export PATH=$JAVA_HOME/bin: source /etc/profile二、集群搭建
解压kafka_2.13-3.0.0.tgz到任意安装目录，在解压出来的目录中进入到config目录进行zookeeper.properties文件混合server.properties文件的配置
tar -zxvf kafka_2.13-3.0.0.tgz cd kafka_2.13-3.0.0/config vim zookeeper.properties
进行如上图设置
dataDir： 存储快照文件snapshot的目录。默认情况下，事务日志也会存储在这里。建议同时配置参数dataLogDir, 事务日志的写性能直接影响zk性能。
datLogDir： 事务日志输出目录。尽量给事务日志的输出配置单独的磁盘或是挂载点，这将极大的提升ZK性能。
tickTime： ZK中的一个时间单元。ZK中所有时间都是以这个时间单元为基础，进行整数倍配置的。例如，session的最小超时时间是2*tickTime。
syncLimit： 在运行过程中，Leader负责与ZK集群中所有机器进行通信，例如通过一些心跳检测机制，来检测机器的存活状态。如果L发出心跳包在syncLimit之后，还没有从F那里收到响应，那么就认为这个F已经不在线了。注意：不要把这个参数设置得过大，否则可能会掩盖一些问题。
quorumListenOnAllIPs：如果是在云服务器上进行集群搭建，此参数最好带上。
后面的三个server分别是三台服务器的IP地址配置，2888端口是zookeeper内部通信端口，3888是zookeeper集群外部通信端口。
server.***：对应server.properties中的broker.id值，然后需要在zookeeper存储目录中添加 myid文件，文件中的内容就是broker.id值。
以下是kafka的server.properties文件配置
borker.id：默认为0，最小为0
advertiset.listeners：对外发布的监听地址，此处应填写为本机IP地址(如果是云服务器就是云服务器IP，如果是物理机就是物理机IP)。
log.dirs：日志目录
zookeeper.connect：zookeeper集群地址，三台机器的IP地址加监听端口
delete.topic.enable：是否能删除topic
以上为全部配置，我是在两台腾讯云服务器和一台阿里云服务器上进行搭建的。
三、启动kafka集群启动三台集群上的kafka之前需要先分别启动三台机器上的zookeeper，当zookeeper集群启动完成后再分别启动kafka
#进入到kafka的bin目录执行命令 cd home/kafka_2.13-3.0.0/bin/ nohup ./zookeeper-server-start.sh ../config/zookeeper.properties >> ../logs/zookeeper.log 2>&1 & #启动kafka命令 nohup ./kafka-server-start.sh ../config/server.properties >> ../logs/kafka.log 2>&1 &
这里为什么使用三台服务器进行机器搭建，因为zookeeper的集群规则是2N+1
1.zookeeper集群的写操作，由leader节点负责，它会把通知所有节进行写入操作，只有收到半数以上节点的成功反馈，才算成功。如果是部署2个节点的话，那就必须都成功。
2.zookeeper的选举策略也是需要半数以上的节点同意才能当选leader，如果是偶数节点可能导致票数相同的情况
3.只有当半数以上的节点存活时 zookeeper集群才能对外服务，维持正常状态，如果是2个节点，只要其中一个挂掉，那么剩下的1个并不满足半数以上规则。
