一、准备工作
centos6.8和jvm需要准备64位的，如果为32位，服务启动的时候报java.lang.OutOfMemoryError: Map failed
的错误。
链接：http://pan.baidu.com/s/1pLbj8T1 密码：f4h0
centos6.8 64位
jdk-8u121-linux-x64.tar.gz
kafka_2.12-0.10.2.1.tgz
zookeeper-3.4.5.tar.gz
虚拟机资源有限，所以3台zookeeper服务器和3台kafka服务器都装在了192.168.137.176上面。
二、安装JAVA
在CentOS6.8安装完以后，系统默认安装了open jdk1.7.0，我们在这里选择安装自己的jdk1.8.0 64位。
(1)tar -zxvf /home/test/jdk-8u121-linux-x64.tar.gz
mv jdk1.8.0_121 /usr/local/jdk1.8.0
在/usr/local 路径下面多了一个 jdk1.8.0文件夹。
(2)配置环境变量
在Linux系统中，一般安装软件的配置信息都是存放在 /etc/profile 文件里面。
vim /etc/profile加入下面设置
export JAVA_HOME=/usr/local/jdk1.8.0
export JRE_HOME=${JAVA_HOME}/jre
export PATH=.:${JAVA_HOME}/bin:$PATH
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
(3)source /etc/profile // 这条命令是让配置马上生效
(4)输入 java -version ，结果很遗憾，还是之前CentOS为我们安装好的jdk，说明我们的配置没有生效，
所以此时我们若要使用自己的jdk来替换系统自带的，还需要进行如下设置：
update-alternatives --install /usr/bin/java java /usr/local/jdk1.8.0/bin/java 50
update-alternatives --config java 选择需要的版本
(5)查看链接情况
输入 ll /usr/bin/java
输出 lrwxrwxrwx. 1 root root 22 Jun 2 09:25 /usr/bin/java -> /etc/alternatives/java
输入 ll /etc/alternatives/java
输出 lrwxrwxrwx. 1 root root 26 Jun 2 09:25 /etc/alternatives/java -> /usr/local/jdk1.8.0/bin/java
三、安装zookeeper服务器集群
zookeeper服务器集群规模不小于3个节点。
1.zookeeper服务安装
进行解压： tar -zxvf zookeeper-3.4.5.tar.gz
重命名： mv zookeeper-3.4.5 /usr/local/zookeeper
2.修改环境变量
vi /etc/profile
export ZOOKEEPER_HOME=/usr/local/zookeeper
export PATH=.:${ZOOKEEPER_HOME}/bin:${JAVA_HOME}/bin:$PATH
刷新： source /etc/profile
3.zookeeper配置文件准备
cd /usr/local/zookeeper/conf
cp zoo_sample.cfg zoo1.cfg
cp zoo_sample.cfg zoo2.cfg
cp zoo_sample.cfg zoo3.cfg
zoo1.cfg配置如下
clientPort=2181
dataDir=/usr/local/zookeeper/data/zoo1
tickTime=2000
initLimit=10
syncLimit=5
server.0=127.0.0.1:2888:3888
server.1=127.0.0.1:2889:3889
server.2=127.0.0.1:2890:3890
zoo2.cfg配置如下
clientPort=2182
dataDir=/usr/local/zookeeper/data/zoo2
tickTime=2000
initLimit=10
syncLimit=5
server.0=127.0.0.1:2888:3888
server.1=127.0.0.1:2889:3889
server.2=127.0.0.1:2890:3890
zoo3.cfg配置如下
clientPort=2183
dataDir=/usr/local/zookeeper/data/zoo3
tickTime=2000
initLimit=10
syncLimit=5
server.0=127.0.0.1:2888:3888
server.1=127.0.0.1:2889:3889
server.2=127.0.0.1:2890:3890
4.创建datadir和服务器标识文件
cd /usr/local/zookeeper/
创建文件夹：mkdir -p data/zoo1 data/zoo2 data/zoo3
cd data/zoo1
创建文件myid并填写内容为0：vi myid (内容为服务器标识 ： 0)
其他两个服务器的值修改为1和2
5.启动zookeeper
zkServer.sh start /usr/local/zookeeper/conf/zoo1.cfg
zkServer.sh start /usr/local/zookeeper/conf/zoo2.cfg
zkServer.sh start /usr/local/zookeeper/conf/zoo3.cfg
查询状态：(一个leader和俩个follower)
zkServer.sh status /usr/local/zookeeper/conf/zoo1.cfg
zkServer.sh status /usr/local/zookeeper/conf/zoo2.cfg
zkServer.sh status /usr/local/zookeeper/conf/zoo3.cfg
停止服务：zkServer.sh stop
启动验证：输入jps，存在 3个QuorumPeerMain 进程的话，就说明 Zookeeper 启动成功了。
四、安装kafka集群
1.kafka服务安装
tar -zxvf kafka_2.12-0.10.2.1.tgz
mv kafka_2.12-0.10.2.1 /usr/local/kafka
2.配置文件准备
config/server.properties参数说明
cd /usr/local/kafka/config/
broker.id=0 #当前机器在集群中的唯一标识，和zookeeper的myid性质一样
listeners = PLAINTEXT://your.host.name:9092#当前kafka对外提供服务的端口默认是9092
num.network.threads=3 #borker进行网络处理的线程数
num.io.threads=8 #borker进行I/O处理的线程数
log.dirs=/usr/local/kafka/kafkalogs1/ #消息存放的目录，可以配置为逗号分割的表达式，上面的num.io.threads要大于这个目录的个数。如果配置多个目录，新创建topic的消息持久化的地方是分区数最少的那一个。
socket.send.buffer.bytes=102400 #发送缓冲区buffer大小，数据不是一下子就发送的，先会存储到缓冲区，到达一定的大小后再发送以提高性能。
socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小，当数据到达一定大小后再序列化到磁盘。
socket.request.max.bytes=104857600 #向kafka请求消息或者向kafka发送消息的请求的最大数，这个值不能超过java的堆栈大小。
num.partitions=1 #默认的分区数，一个topic默认1个分区数
log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天
message.max.byte=5242880 #消息保存的最大值5M
default.replication.factor=2 #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务
replica.fetch.max.bytes=5242880 #取消息的最大值
log.segment.bytes=1073741824 #kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间，到目录查看是否有过期的消息如果有，删除
log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能
zookeeper.connect=localhost:2181 #设置zookeeper的连接端口
配置文件修改
broker.id=0 #每台服务器的broker.id都不能相同
listeners=PLAINTEXT://192.168.137.176:9092
log.dirs=/usr/local/kafka/kafkalogs1/
#增加下面设置
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
#设置zookeeper的连接端口
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
其他2个服务器配置文件
cp server.properties server-1.properties
cp server.properties server-2.properties
broker.id=1 #每台服务器的broker.id都不能相同
listeners=PLAINTEXT://192.168.137.176:9093
log.dirs=/usr/local/kafka/kafkalogs2/
#增加下面设置
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
#设置zookeeper的连接端口
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
broker.id=2 #每台服务器的broker.id都不能相同
listeners=PLAINTEXT://192.168.137.176:9094
log.dirs=/usr/local/kafka/kafkalogs3/
#增加下面设置
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
#设置zookeeper的连接端口
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
3.服务器启动
cd /usr/local/kafka/
setsid bin/kafka-server-start.sh config/server.properties &
setsid bin/kafka-server-start.sh config/server-1.properties &
setsid bin/kafka-server-start.sh config/server-2.properties &
setsid作用是让服务在linux客户端退出时依然启动。
4.服务器验证
输入jps，输出如下：3个zookeeper服务，3个kafka服务。
1794 Kafka
2069 Kafka
1253 QuorumPeerMain
1321 QuorumPeerMain
3658 Jps
1275 QuorumPeerMain
1517 Kafka
五、测试kafka集群
1.创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
2.查询topic状态
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
3.发布消息
bin/kafka-console-producer.sh --broker-list 192.168.137.176:9092 --topic my-replicated-topic
4.接收消息
bin/kafka-console-consumer.sh --bootstrap-server 192.168.137.176:9092 --from-beginning --topic my-replicated-topic
5.查看topic
bin/kafka-topics.sh --list --zookeeper localhost:2181
kafka集群的创建已经完成。