首先提到kafka集群，不免的会与zookeeper联系在一起 。
ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务。它是集群的管理者，监视着集群中各个节点的状态根据节点提交的反馈进行下一步合理操作。最终，将简单易用的接口和性能高效、功能稳定的系统提供给用户。
kafka 通俗的来说是消息中间件 与之类似或是经常使用如：ActiveMg、RabbitMq等 由于是展示实战的问题，在这就不做过多的介绍（自己查缺补漏）。
总结：搭建kafka集群 ，首先搭建zookeeper主从copy，再做kafka集群。有人问我，用单一节点的zookeeper不行吗？答案是可行的 。但是你想一下，在搭建kafka集群的同时你的硬件设备是支持，搭建zookeeper集群的，你为什么不把集群做的更稳定一些呢（将其放到一个zookeeper上。如果zookeeper挂掉怎么办）。
综上所述开干：
我用的服务器为：centos7（个人觉得稳定一些）
zookeeper集群的搭建：
http://zookeeper.apache.org（zookeeper官方网址）
（1）安装zookeeper
推荐安装方法 ：wget http://mirrors.hust.edu.cn/apache/zookeeper
官方提供的下载路径供给选择
http://mirror.bit.edu.cn/apache/zookeeper/

http://mirrors.hust.edu.cn/apache/zookeeper/

http://mirrors.shu.edu.cn/apache/zookeeper/

http://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/

（2）下载完成后，解压安装
tar -zxvf zookeeper-3.4.10.tar.gz
（3）修改配置文件
下载完成找到 zookeeper-3.4.10/conf/ zoo_sample.cfg
将 zoo_sample.cfg copy一份 （cp zoo_sample.cfg zoo.cfg）
寻找修改文件的路径 zookeeper-3.4.10/conf/zoo.cfg
tickTime=2000（Zookeeper的时间单元。Zookeeper中所有时间都是以这个时间单元的整数倍去配置的。例如，session的最小超时时间是2*tickTime。（单位：毫秒））
initLimit=10（Observer和Follower启动时，从Leader同步最新数据时，Leader允许initLimit * tickTime的时间内完成。如果同步的数据量很大，可以相应的把这个值设置的大一些。）
syncLimit=5（表示Follower和Observer与Leader交互时的最大等待时间，只不过是在与leader同步完毕之后，进入正常请求转发或ping等消息交互时的超时时间。）
dataDir=/data/zookeeper/data（用于存放内存数据快照的文件夹，同时用于集群的myid文件也存在这个文件夹里）
dataLogDir=/data/zookeeper/log（事务日志写入该配置指定的目录，而不是“ dataDir ”所指定的目录。这将允许使用一个专用的日志设备并且帮助我们避免日志和快照之间的竞争）
clientPort=2181（服务的监听端口）
server.1=10.0.0.20:6888:5888（表示 Flower 跟 Leader的通信端口，简称服务端内部通信的端口（默认6888），表示 是选举端口（默认是5888））
server.2=10.0.0.5:6888:5888 server.3=10.0.0.199:6888:5888

例如

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data //注意要创建对应的目录
dataLogDir=/data/zookeeper/log //注意要创建对应的目录
clientPort=2181
server.1=10.0.0.20:6888:5888 //注意 server 后面是 myid 要与myid文件中的值一致
server.2=10.0.0.5:6888:5888
server.3=10.0.0.199:6888:5888

注意：server 后面是 myid 要与myid文件中的值一致
（4）配置其余两台服务器上的zookeeper
与之一样 直接按第一台配置就好。
（5）启动配置好的三台服务器
sh zookeeper-3.4.10/bin/zkServer.sh start(启动)
sh zookeeper-3.4.10/bin/zkServer.sh stop （关闭）
启动完成后 查看一下进程
jps

表示有zookeeper的进程
或者
ps -ef |grep “zookeeper”
（5）zookeeper集群测试
sh zookeeper-3.4.10/bin/zkCli.sh
看一下 zookeeper客户端是否能正常访问

表示 客户端能够正常访问
（6）测试zookeeper的主从copy
sh zookeeper-3.4.10/bin/zkServer.sh status

看到这就代表zookeeper主从copy 就成功啦

二、kafka集群
http://kafka.apache.org（官方文档）
（1）安装kafka
wget http://mirrors.hust.edu.cn/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz
提供的路径：
http://mirror.bit.edu.cn/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz

http://mirrors.hust.edu.cn/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz

http://mirrors.shu.edu.cn/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz

http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz
（2）下载完成后，解压安装
tar -zxvf kafka_2.11-2.0.0.tgz
（3）修改配置文件
找到修改文件 kafka_2.11-2.0.0/config/server.porperties

host.name=地址  //broker的主机地址，若是设置了，那么会绑定到这个地址上，若是没有，会绑定到所有的接口上，并将其中之一发送到ZK，一般不设置
 #在log.retention.hours=168 //下面新增下面三项
 message.max.byte=5242880 //表示消息体的最大大小，单位是字节
 default.replication.factor=3 //消息备份数目
 replica.fetch.max.bytes=5242880 //replicas每次获取数据的最大大小
 #设置zookeeper的连接端口
 zookeeper.connect=地址:端口,地址:端口,地址:端口

注意： broker.id 是唯一的
如果出现 kafka 启动后闪退 ，那么你要检查一下 server.porperties中broker.id 是否唯一 。
解决方法 ： 修改 server.porperties 的broker.id 然后删除 log.dirs=/tmp/kafka-logs
下的所有日志 重启一下

（3）创建一个producer（用于发布消息）
sh bin/kafka-console-producer.sh –broker-list 地址:9092 –topic xxx

（4）查看主题
bin/kafka-topics.sh –describe –zookeeper 地址:2181
查看是否有创建的主题 xxx
（5）创建一个 consumer
sh bin/kafka-console-consumer.sh –zookeeper 地址:2181 –topic xxx t –from-beginning
查看生产者发布的消息有没有发布成功 ，消费者是否能接受到

如果上述完成那么kafka集群配置成功！

环境搭载成功！
如果遇到问题 欢迎联系作者 ： 13071880500