因为Kafka集群是把状态信息保存在Zookeeper中的，并且Kafka的动态扩容是通过Zookeeper来实现的，所以需要优先搭建Zookeerper集群，建立分布式状态管理。开始准备环境，搭建集群：
zookeeper是基于Java环境开发的所以需要先安装Java 然后这里使用的zookeeper安装包版本为zookeeper-3.4.14，Kafka的安装包版本为kafka_2.11-2.2.0。
AMQP协议：Advanced Message Queuing Protocol （高级消息队列协议）是一个标准开放的应用层的消息中间件协议。AMQP定义了通过网络发送的字节流的数据格式。因此兼容性非常好，任何实现AMQP协议的程序都可以和与AMQP协议兼容的其他程序交互，可以很容易做到跨语言，跨平台
server1:192.168.177.251
server2:192.168.177.252
server3:192.168.177.132
安装之前先检查一下系统有没有自带open-jdk
命令：
rpm -qa |grep java
rpm -qa |grep jdk
如果没有输入信息表示没有安装。
检索1.8的列表
yum list java-1.8*
安装1.8.0的所有文件
yum install java-1.8.0-openjdk* -y
使用命令检查是否安装成功
java -version
cat /etc/hosts
192.168.177.251 kafka01
192.168.177.252 kafka02
192.168.177.133 kafka03
1.关闭防火墙。selinux
systemctl stop firewalld
setenforce 0
安装zookeeper && kafka
wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.0/kafka_2.11-2.2.0.tgz
放到/usr/local下
[root@192 ~]# cd /usr/local/zookeeper/conf/ [root@192 conf]# mv zoo_sample.cfg zoo.cfg [root@192 conf]# vim zoo.cfg
tickTime=2000 #zookeeper服务器之间的心跳时间。
initLimit=10 #zookeeper的最大连接失败时间
syncLimit=5 #zookeeper的同步通信时间
dataDir=/tmp/zookeeper #zookeeper的存放快照日志的绝对路径
dataLogDir=/usr/local/zookeeper/zkdatalog #zookeeper的存放事物日志的绝对路径
clientPort=2181 #zookeeper与客户端的连接端口
server.1=192.168.177.251:2888:3888 server.2=192.168.177.252:2888:3888 server.3=192.168.177.132:2888:3888 ###服务器及其编号，服务器IP地址，通信端口，选举端口
创建myid文件
[root@192 conf]# mkdir /tmp/zookeeper [root@192 conf]# echo "1">/tmp/zookeeper/myid #另外两台节点也要做。重复以上步骤
启动zookeeper
/usr/local/zookeeper/bin/zkServer.sh start 启动 /usr/local/zookeeper/bin/zkServer.sh status 查看状态
Mode: leader为主节点，Mode: follower为从节点，zk集群一般只有一个leader，多个follower，主一般是响应客户端的读写请求，而从主同步数据，当主挂掉之后就会从follower里投票选举一个leader出来。
到此，zookeeper集群搭建结束，接下来基于zookeeper搭建kafka集群：
Kafka的基本概念：
主题：Topic特指Kafka处理的消息源（feeds of messages）的不同分类。
分区：Partition Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。
Message：消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。
Producers：消息的数据生产者，向Kafka的一个topic发布消息的过程叫做producers。
Consumers：消息的数据消费者，订阅topics并处理其发布的消息的过程叫做consumers。
Broker：缓存代理，Kafka集群中的一台或多台服务器统称为broker，这里用的是AMQP协议。
1. 修改配置文件
cd /usr/local/kafka/config vim server.properties broker.id=0 #后面的数字唯一的。不可重复 advertised.listeners=PLAINTEXT://kafka01:9092 log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天 message.max.byte=5242880 #消息保存的最大值5M default.replication.factor=2 #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务 replica.fetch.max.bytes=5242880 #取消息的最大直接数 zookeeper.connect=192.168.177.251:2181,192.168.177.252:2181,192.168.177.132:2181
注意：先现在把主机名字也给改了
hostname kafka01
2.启动kafka集群
[root@192 bin]# /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
3，检测端口 kafka 端口为9092
netstat -anpt |grep :9092
4.测试
创建topic
[root@192 config]# cd /usr/local/kafka/bin/ [root@192 bin]# ./kafka-topics.sh --create --zookeeper 192.168.177.132:2181 --replication-factor 2 --partitions 3 --topic wg01
#--replication-factor 2 复制两份
#--partitions 3 创建三个分区
#--topic wg007 主题为wg007
5、模拟一个producter：
./kafka-console-producer.sh --broker-list 192.168.177.132:9092 --topic wg01
6、模拟一个consumer：
./kafka-console-consumer.sh --bootstrap-server 192.168.177.132:9092 --topic wg01 --from-beginning
7、查看topic：
./kafka-topics.sh --list --zookeeper 192.168.177.132:2181