zookeeper集群和kafka集群搭建
kafka集群是把状态保存到zookeeper集群上的，所有要创建kafka集群需先创建zookeeper集群
本次实验环境：
操作系统centos7
主机ip：192.168.1.9、192.168.1.10、192.168.1.11
注：Zookeeper集群的工作是超过半数才能对外提供服务，3台中超过两台则为超过半数，允许1台挂掉，集群可使用2*n+1台搭建。
1、安装zookeeper集群
选择Zookeeper的稳定版本Zookeeper 3.4.6版本
安装java环境
yum -y install java-1.8.0-openjdk*
下载zookeeper
本次将zookeeper安装到/opt/zookeeper下
mkdir /opt/zookeeper
mkdir /opt/zookeeper/zkdata
mkdir /opt/zookeeper/zkdatalog
wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz（若连接失效则自行下载）
tar -zxvf zookeeper-3.4.6.tar.gz
mv zookeeper-3.4.6/* /opt/zookeeper/
修改zookeeper配置文件
cd /opt/zookeeper/conf
cp zoo_sample.cfg zoo.cfg
vi zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/zookeeper/zkdata
dataLogDir=/opt/zookeeper/zkdatalog
clientPort=2181
server.1=192.168.1.9:2888:3888
server.2=192.168.1.10:2888:3888
server.3=192.168.1.11:2888:3888
#server.1 这个1是服务器的标识也可以是其他的数字， 表示这个是第几号服务器，用来标识服务器，这个标识要写到快照目录下面myid文件里
#192.168.1.9为集群里的IP地址，第一个端口是master和slave之间的通信端口，默认是2888，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是3888
配置文件详解：
#tickTime：
这个时间是作为Zookeeper服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一个心跳
#initLimit：
这个配置项是用来配置Zookeeper接受客户端（这里所说的客户端不是用户连接Zookeeper服务器的客户端，而是Zookeeper 服务器集群中连接到Leader的Follower服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 5个心跳的时间（也就是tickTime）长度后Zookeeper服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是5*2000=10秒
#syncLimit：
这个配置项标识Leader与Follower之间发送消息，请求和应答时间长度，最长不能超过多少个tickTime 的时间长度，总的时间长度就是5*2000=10秒
#dataDir：
快照日志的存储路径
#dataLogDir：
事物日志的存储路径，如果不配置这个那么事物日志会默认存储到dataDir制定的目录，这样会严重影响zk的性能，当zk 吞吐量较大的时候，产生的事物日志、快照日志太多
#clientPort：
这个端口就是客户端连接Zookeeper服务器的端口，Zookeeper会监听这个端口，接受客户端的访问请求
创建myid文件
三台主机根据自己的ip创建不同的myid文件
#server1
echo "1" > /opt/zookeeper/zkdata/myid
#server2
echo "2" > /opt/zookeeper/zkdata/myid
#server3
echo "3" > /opt/zookeeper/zkdata/myid
重要配置说明
1、myid文件和server.myid在快照目录下存放的标识本台服务器的文件，他是整个zk集群用来发现彼此的一个重要标识。
2、zoo.cfg 文件是zookeeper配置文件在conf目录里。
3、log4j.properties文件是zk的日志输出文件在conf目录里用java写的程序基本上有个共同点日志都用log4j，来进行管理。
4、zkEnv.sh和zkServer.sh文件
zkServer.sh 主的管理程序文件
zkEnv.sh 是主要配置，zookeeper集群启动时配置环境变量的文件
5、zookeeper不会主动的清除旧的快照和日志文件，这个是操作者的责任。一下为清除文件的实例脚本：
#!/bin/bash
dataDir=/opt/zookeeper/zkdata/version-2
dataLogDir=/opt/zookeeper/zkdatalog/version-2
count=66
count=$[$count+1]
ls -t $dataLogDir/log.* | tail -n +$count | xargs rm -f
ls -t $dataDir/snapshot.* | tail -n +$count | xargs rm -f
#以上这个脚本定义了删除对应两个目录中的文件，保留最新的66个文件，可以将他写到crontab 中，设置为每天凌晨2点执行一次就可以了。
启动并查看服务
启动服务
cd /opt/zookeeper/bin
./zkServer.sh start
查看服务状态
./zkServer.sh status
JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.6/bin/../conf/zoo.cfg #配置文件
Mode: follower #他是否为领导
2、kafka集群搭建
搭建环境
1、linux主机大于等于3。
2、搭建好的zookeeper集群。
3、软件版本kafka_2.11-0.9.0.1.tgz
创建目录并下载软件
mkdir /opt/kafka
mkdir /opt/kafka/kafkalogs
wget http://apache.opencas.org/kafka/0.9.0.1/kafka_2.11-0.9.0.1.tgz（连接不可用则自行下载）
tar -zxvf kafka_2.11-0.9.0.1.tgz
mv kafka_2.11-0.9.0.1/* /opt/kafka/
修改kafka配置文件
cd /opt/kafka/config
cp server.properties server.properties.bak
vi server.properties 只列出修改部分
broker.id=0 #每台服务器的broker.id都不能相同
listeners=PLAINTEXT://192.168.1.9:9092 #三台分别写本机ip
log.dirs=/opt/kafka/kafkalogs/
#在log.retention.hours=168下面新增下面三项
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
#设置zookeeper的连接端口
zookeeper.connect=192.168.1.9:2181,192.168.1.10:2181,192.168.1.11:2181
配置文件详解：
broker.id=0 #当前机器在集群中的唯一标识，和zookeeper的myid性质一样
port=9092 #当前kafka对外提供服务的端口默认是9092
host.name=192.168.1.9 #这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。
num.network.threads=3 #这个是borker进行网络处理的线程数
num.io.threads=8 #这个是borker进行I/O处理的线程数
log.dirs=/opt/kafka/kafkalogs/ #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的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.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除
log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能
zookeeper.connect=192.168.7.100:12181,192.168.7.101:12181,192.168.7.107:1218 #设置zookeeper的连接
注：配置文件中advertised.listeners和listeners的区别
listeners是kafka真正bind的地址，advertised.listeners是暴露给外部的listeners，如果没有设置，会用listeners
启动Kafka集群并测试
启动服务
cd /opt/kafka/bin
./kafka-server-start.sh -daemon ../config/server.properties
jps -l 查看是否启动成功
创建Topic来验证是否创建成功
创建topic
./kafka-topics.sh --create --zookeeper 192.168.1.9:2181 --replication-factor 2 --partitions 1 --topic shuaige
--replication-factor 2 #复制两份
--partitions 1 #创建1个分区
--topic #主题为shuaige
创建发布者
./kafka-console-producer.sh --broker-list 192.168.1.9:9092 --topic shuaige
创建订阅者
./kafka-console-consumer.sh --zookeeper 192.168.1.9:2181 --topic shuaige --from-beginning
测试在发布者那里发布消息看看订阅者那里是否能正常收到
其它说明
默认kafka的日志是保存在/opt/kafka/logs目录下
server.log #kafka的运行日志
state-change.log #kafka他是用zookeeper来保存状态，所以他可能会进行切换，切换的日志就保存在这里
controller.log #kafka选择一个节点作为“controller”,当发现有节点down掉的时候它负责在永久分区的所有节点中选择新的leader,这使得Kafka可以批量的高效的管理所有分区节点的主从关系。如果controller down掉了，活着的节点中的一个会备切换为新的controller
使用KafkaOffsetMonitor监控kafka集群
下载KafkaOffsetMonitor-assembly-0.2.0.jar
KafkaOffsetMonitor托管在Github上，可以通过Github下载。
下载地址：https://github.com/quantifind/KafkaOffsetMonitor/releases
创建监控脚本
mkdir /opt/kafkamonitor
mv KafkaOffsetMonitor-assembly-0.2.0.jar /opt/kafkamonitor/
vi kafka-monitor-start.sh
#! /bin/bash
java -cp KafkaOffsetMonitor-assembly-0.2.0.jar \
com.quantifind.kafka.offsetapp.OffsetGetterWeb \
--zk 192.168.1.9:2181 \
--port 18089 \
--refresh 10.seconds \
--retain 1.days
chmod +x kafka-monitor-start.sh
启动监控空查看
nohup /data/KafkaMonitor/kafka-monitor-start.sh&
然后在本地用浏览器访问192.168.1.9:18089，可查看监控
