分布式集群如何实现使用Kafka多节点搭建？相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。
多节点分布式集群结构如下图所示：
为了方便查阅，本篇将和上一篇一样从零开始一步一步进行集群搭建。
一、安装Jdk
具体安装步骤可参考linux安装jdk。
二、安装与配置zookeeper
下载地址：https://www-us.apache.org/dist/zookeeper/stable/
下载二进制压缩包zookeeper-3.4.14.tar.gz，然后上传到linux服务器指定目录下，本次上传目录为/software，然后执行如下命令安装：
cd /software
tar -zxvf zookeeper-3.4.14.tar.gz
mv zookeeper-3.4.14 /usr/local/zookeeper
cd /usr/local/zookeeper/conf
mv zoo_sample.cfg zoo1.cfg
编辑zoo1.cfg,配置相关参数如下：
tickTime=2000
initLimit=5
syncLimit=2
dataDir=/usr/local/zookeeper/data/zookeeper1
clientPort=2181
server.1=192.168.184.128:2888:3888
server.2=192.168.184.128:2889:3889
server.3=192.168.184.128:2890:3890
其中：
tickTime：Zookeeper最小的时间单位，用于丈量心跳和超时时间，一般设置默认值2秒；
initLimit：指定follower节点初始时连接leader节点的较大tick此处，设置为5，表示follower必须在5xtickTime即10秒内连接上leader,否则视为超时；
syncLimit：设定follower节点与leader节点进行同步的较大时间，设置为2，表示较大时间为2xtickTime即4秒时间；
dataDir：Zookeeper会在内存中保存系统快照，并定期写入该路径指定的文件夹中，生产环境需要特别注意该文件夹的磁盘占用情况；
clientPort：Zookeeper监听客户端连接的端口号，默认为2181，同一服务器上不同实例之间应该有所区别；
server.X=host:port1:port2:此处X的取值范围在1~255之间，必须是全局的且和myid文件中的数字对应（myid文件后面说明）,host是各个节点的主机名，port1通常是2888，用于使follower节点连接leader节点，port2通常是3888，用于leader选举，zookeeper在不同服务器上的时候，不同zookeeper服务器的端口号可以重复，在同一台服务器上的时候需要有所区别。
1.配置zoo.cfg文件
单节点安装zookeeper的时候，仅有一份zoo.cfg文件，多节点安装的时候，每个zookeeper服务器就应该有一个zoo.cfg配置文件。如果在一台服务器安装zookeeper多实例集群，则需要在conf目录下分别配置每个实例的zoo.cfg，同时创建每个zookeeper实例自己的数据存储目录。本次在一台服务器上配置多个zookeeper实例，执行如下命令创建数据存储目录并复制配置文件：
mkdir -p /usr/local/zookeeper/data/zookeeper1
mkdir -p /usr/local/zookeeper/data/zookeeper2
mkdir -p /usr/local/zookeeper/data/zookeeper3
cd /usr/local/zookeeper/conf/
cp zoo1.cfg zoo2.cfg
cp zoo1.cfg zoo3.cfg
复制后分别修改zoo2.cfg,zoo3.cfg中的配置，修改后的配置如下：
zoo1.cfg的配置如下:
zoo2.cfg的配置如下:
zoo3.cfg中的配置如下：
2.myid文件创建与配置
前面提到zoo.cfg文件中的server.X中的X应该与myid中的数字相对应。除此之外，myid文件必须存放在每个zookeeper实例的data目录下，对应本次安装应该位于/usr/local/zookeeper/data/zookeeper1，2,3目录下，执行如下命令进行配置：
echo '1' > /usr/local/zookeeper/data/zookeeper1/myid
echo '2' > /usr/local/zookeeper/data/zookeeper2/myid
echo '3' > /usr/local/zookeeper/data/zookeeper3/myid
3.启动zookeeper服务器
使用如下命令启动zookeeper集群:
cd /usr/local/zookeeper/bin/
./zkServer.sh start ../conf/zoo1.cfg
./zkServer.sh start ../conf/zoo2.cfg
./zkServer.sh start ../conf/zoo3.cfg
启动后，使用如下命令查看集群状态:
cd /usr/local/zookeeper/bin/
./zkServer.sh status ../conf/zoo1.cfg./zkServer.sh status ../conf/zoo2.cfg./zkServer.sh status ../conf/zoo3.cfg
回显信息如下：
可以看到有两个follower节点，一个leader节点。
三、安装与配置kafka集群
下载地址：http://kafka.apache.org/downloads.html
1.数据目录和配置文件创建
目前新版本是2.2.0,本次下载2.1.1版本的安装包，然后上传压缩包到服务器指定目录，本次上传目录为/software，然后执行以下命令进行安装:
tar -zxvf kafka_2.12-2.1.1.tgz
mv kafka_2.12-2.1.1 /usr/local/kafka
mkdir -p /usr/local/kafka/logs/kafka1
mkdir -p /usr/local/kafka/logs/kafka2
mkdir -p /usr/local/kafka/logs/kafka3
cd /usr/local/kafka/config/
mv server.properties server1.properties
通过执行上面的命令，我们在/usr/local/kafka/logs文件夹中创建了kafka1,kafka2,kafka3三个文件夹用于存放三个kafka实例的数据，同时将/usr/local/kafka/config/文件夹下的server.properties重命名为server1.properties用于配置kafka的第一个实例。
2.配置属性文件
接下来配置server1.properties文件，主要配置参数如下：
broker.id=1：设置kafka broker的id，本次分别为1,2,3；
delete.topic.enable=true:开启删除topic的开关；
listeners=PLAINTEXT://192.168.184.128:9092：设置kafka的监听地址和端口号，本次分别设置为9092,9093,9094；
log.dirs=/usr/local/kafka/logs/kafka1：设置kafka日志数据存储路径；
zookeeper.connect=192.168.184.128:2181,192.168.184.128:2182,192.168.184.128:2183：设置kafka连接的zookeeper访问地址,集群环境需要配置所有zookeeper的访问地址;
unclean.leader.election.enable=false：为true则代表允许选用非isr列表的副本作为leader，那么此时就意味着数据可能丢失，为false的话，则表示不允许，直接抛出NoReplicaOnlineException异常，造成leader副本选举失败。
zookeeper.connection.timeout.ms=6000：设置连接zookeeper服务器超时时间为6秒。
配置完成后，复制server1.properties两份分别用于配置kafka的第二个，第三个节点：
复制代码 代码如下:
