本文主要介绍 Kafka 集群搭建，本文的内容以 Linux CentOS7 作为安装演示，其他 Linux 系列的操作系统也可以根据情况参考本文的内容。具体的操作流程如下：

前提

kafka安装的前提需要有jdk环境，还需要涉及ZooKeeper，没有安装的朋友可以先安装 jdk以及zookeeper。

为了节省大家的时间，我已准备好了安装包，可通过以下方式提取：

链接：https://pan.baidu.com/s/1NK-jrlJxbrdiNFZeGqpM-A
提取码：hra5

压缩包内容：

一.安装JDK

1.通过以下命令解压压缩包

tar -zxvf jdk-8u181-linux-x64.tar.gz

2.配置 JDK 的环境变量。修改/etc/profile 文件并向其中添加如下配置：
修改profile文件

vim /etc/profile

在profile文件添加以下环境

export JAVA_HOME=/opt/kafka/jdk1.8.0_181
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

3.执行 source /etc/profile 命令使配置生效。最后可以通过 java –version命令验证 JDK 是否已经安装配置成功。如果安装配置成功，则会正确显示出 JDK 的版本信息，参考如下：

二.搭建Zookeeper集群

1.通过以下命令解压压缩包

tar -zxvf apache-zookeeper-3.5.7-bin.tar.gzjdk-8u181-linux-x64.tar.gz

2.向/etc/profile 配置文件中添加如下内容：
修改profile文件

vim /etc/profile

配置文件中添加如下内容：

export ZOOKEEPER_HOME=/opt/kafka/apache-zookeeper-3.5.7-bin
export PATH=$PATH:$ZOOKEEPER_HOME/bin

再执行 source /etc/profile 命令使配置生效。

3.修改 ZooKeeper 的配置文件。首先进入$ZOOKEEPER_HOME/conf目录，并将zoo_sample.cfg 文件修改为 zoo.cfg：
cp zoo_sample.cfg zoo.cfg

4.修改apache-zookeeper-3.5.7-bin目录名称
mv apache-zookeeper-3.5.7-bin zookeeper1
复制三份，搭建集群
cp -r zookeeper1 zookeeper2
cp -r zookeeper1 zookeeper3

5.分别修改zoo.cfg.
进入到cd /opt/kafka/zookeeper_cluster/zookeeper1/conf ,vim zoo.cfg

dataDir:服务器存储快照文件目录。
dataLogDir:服务器存日志文件目录。
server.A=B:C:D
含义：
A:节点在集群中的唯一ID，需要与dataDir/myid文件中的内容保持一致，取值范围为1~255。
B：节点的服务器IP地址。
C：集群中Follower节点与Leader 节点之间通信的端口。
D：当集群中的Follower节点宕机或出现故障时，集群进行重新选举Leader时所使用的端口。

然后分别进入到zookeeper2,zookeeper3,执行步骤5的操作，修改dataDir，dataLogDir到指定目录。

6.在每个zookeeper的 data 目录下创建一个 myid 文件，内容分别是1、2、3 。这个文件就是记录每个服务器的ID。（没有data目录的可以先创建data目录mkdir data）

echo 1 >/opt/kafka/zookeeper_cluster/zookeeper1/data/myid
echo 2 >/opt/kafka/zookeeper_cluster/zookeeper2/data/myid
echo 3 >/opt/kafka/zookeeper_cluster/zookeeper3/data/myid

7.启动zookeeper。分别进入三个zookeeper目录，执行以下命令：
./bin/zkServer.sh start

8.查看zookeeper状态
./bin/zkServer.sh status

可以看到，三个服务都启动成功，其中zookeeper2为leader（领导者），其余两个为follower（跟随者），到此，zookeeper集群已搭建成功。

三.安装kafka并搭建集群

1.解压kafka:

tar -zxvf kafka_2.11-2.4.0.tgz

2.进入解压目录 cd /opt/kafka/kafka_cluster，修改目录名mv kafka_2.11-2.4.0 kafka1

3.复制另外两份，搭建集群
cp -r kafka1 kafka2
cp -r kafka1 kafka3

3.修改配置文件

进入config目录
/opt/kafka/kafka_cluster/kafka1/config
修改server.properties文件
vim server.properties

主要修改以下几点
kafka1 server.properties:

#每个broker在集群中的唯一标识，不能重复
broker.id=0
listeners=PLAINTEXT://10.0.12.16:9092
advertised.listeners=PLAINTEXT://43.139.86.193:9092
#日志存放目录
log.dirs=/opt/kafka/kafka_cluster/kafka1/logs
#zookeeper集群
zookeeper.connect=10.0.12.16:2181,10.0.12.16:2182,10.0.12.16:2183

kafka2 server.properties:

#每个broker在集群中的唯一标识，不能重复
broker.id=1
listeners=PLAINTEXT://10.0.12.16:9093
advertised.listeners=PLAINTEXT://43.139.86.193:9093
#日志存放目录
log.dirs=/opt/kafka/kafka_cluster/kafka2/logs
#zookeeper集群
zookeeper.connect=10.0.12.16:2181,10.0.12.16:2182,10.0.12.16:2183

kafka3 server.properties:

#每个broker在集群中的唯一标识，不能重复
broker.id=2
listeners=PLAINTEXT://10.0.12.16:9094
advertised.listeners=PLAINTEXT://43.139.86.193:9094
#日志存放目录
log.dirs=/opt/kafka/kafka_cluster/kafka3/logs
#zookeeper集群
zookeeper.connect=10.0.12.16:2181,10.0.12.16:2182,10.0.12.16:2183

4.启动kafka服务：

cd /opt/kafka/kafka_cluster/kafka1
bin/kafka-server-start.sh config/server.properties &

cd /opt/kafka/kafka_cluster/kafka2
bin/kafka-server-start.sh config/server.properties &

cd /opt/kafka/kafka_cluster/kafka3
bin/kafka-server-start.sh config/server.properties &

若启动报以下错误：
The Cluster ID *** doesn‘t match stored clusterId Some(***) in meta.properties The broker

解决方案：
进入到kafka_2.11-2.4.0/config/server.properties,找到log.dirs对应路径。

进入/tmp/kafka-logs找到meta.properties文件
vim meta.properties修改cluster.id为错误日志中的ID;
修改broker.id为kafka_2.11-2.4.0/config/server.properties里面的broker.id

修改完成之后再次启动就成功了。

四.验证

1.创建一个topic:

bin/kafka-topics.sh --create --zookeeper 10.0.12.16:2181,10.0.12.16:2182,10.0.12.16:2183 --replication-factor 2 --partitions 2 --topic topic1

2.查看已经创建的Topic信息

bin/kafka-topics.sh --list --zookeeper 10.0.12.16:2181,10.0.12.16:2182,10.0.12.16:2183

3.发送消息

首先多开一个窗口，执行以下命令,做为消费者去接收消息。

bin/kafka-console-consumer.sh --bootstrap-server 43.139.86.193:9092 --topic topic1 --from-beginning

然后执行以下命令发送消息。

bin/kafka-console-producer.sh --broker-list 43.139.86.193:9092 --topic topic1

结果：
发送页面：

接收页面：

可以看到，消息也成功发送，并且被成功接收到。

以上就是本文所有内容，如有不当，欢迎纠正！