准备好三台服务器，集群中的每一台服务器都必须配置好相应的JAAS，修改三台主机的hosts文件，ip地址对应主机名

vim /etc/hosts

192.168.5.161 kafka-cluster-a
192.168.5.162 kafka-cluster-b
192.168.5.163 kafka-cluster-c
需要对每一台主机的Zookeeper进行配置，配置zookeeper集群主机。

vim /usr/local/kafka/config/zookeeper.properties
initLimit=5
syncLimit=5
server.1=kafka-cluster-a:2888:3888
server.2=kafka-cluster-b:2888:3888
server.3=kafka-cluster-c:2888:3888

设置每一个Zookeeper服务主机的mid文件内容

kafka-cluster-a: echo 1 >> /usr/local/zookeeper/myid
kafka-cluster-b: echo 2 >> /usr/data/zookeeper/myid
kafka-cluster-c: echo 3 >> /usr/data/zookeeper/myid

启动所有的Zookeeper服务，后台启动

/usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties > /usr/data/zookeeper/zookeeper.log 2>&1 &

配置所有主机kafka文件：

vim /usr/local/kafka/config/server.properties  
broker.id=0
listeners=SASL_PLAINTEXT://kafka-cluster-x:9095
advertised.listeners=SASL_PLAINTEXT://kafka-cluster-x:9095
num.partitions=3
default.replication.factor=3
zookeeper.connect=kafka-cluster-a:2181,kafka-cluster-b:2181,kafka-cluster-c:2181

broker.id：明确描述每台主机的broker名称，集群中每一台kafka的broker.id都要不相同
listeners：设置监听主机，监听本机
num.partitions：设置几个分区
default.replication.factor：副本数量
zookeeper.connect：Zookeeper地址
如果是直接拷贝的之前的主机还需要删除“/usr/data/kafka/*”下的数据，否则“broker.id”不等于0的主机启动会报错

rm -r /usr/data/kafka/*

随后启动所有主机中的kafka服务

/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties > /usr/data/kafka/kafka.log 2>&1 &

在程序调用kafka集群非常的简单，只需要将集群的地址添加上即可

单节点：private static final String SERVERS = "kafka-server:9095";
集群：private static final String SERVERS = "kafka-cluster-a:9095,kafka-cluster-b:9095,kafka-cluster-c:9095";

该集群设置配置有副本，所谓的副本指定是一个信息是否在其他的主机上进行存储