【环境】
三台虚拟机
192.168.1.224 hadoop2004
192.168.1.225 hadoop2005
192.168.1.226 hadoop2006
确保每台机器上均安装了JDK
确保zookeeper集群（kafka集群的搭建是建立在jdk和zookeeper集群环境之上的）
【1 ：zookeeper集群搭建】
【1.1】解压tar包
在机器192.168.1.224 hadoop2004机器上
tar -xzvf zookeeper3.4.6.tar.gz /usr/local
【1.2】修改配置
cd /usr/local/zookeeper3.4.6/conf
mv zoo_sample.cfg zoo.cfg
Paste_Image.png
配置文件解释
#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 会监听这个端口，接受客户端的访问请求。修改他的端口改大点
【1.3】添加所配置的数据存储目录data
mkdir /usr/local/zookeeper/data
cd data
//在data下创建一个myid文件
touch myid
//在myid文件中编写此机器的编号（这个是server1，则编号就是1）
vim myid （添加1）
注意：这里的myid文件时区分集群中的机器的，所以各个机器上的myid不能重复
【1.4】复制zookeeper目录到hadoop2005和hadoop2006机器上
利用免密码登陆复制文件
scp -r zookeeper root@hadoop02:/usr/local
scp -r zookeeper root@hadoop03:/usr/local
【1.5】
分别修改server2和server3机器目录下、/usr/local/zookeeper/data的myid文件，将文件内容分别该为server编号（2 , 3）
【1.6】
分别在三台机器上启动
./zkServer.sh start（开启）
./zkServer.sh status（重启） 重启可以查看到它的角色状态
【1.7】配置说明
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不会主动的清除旧的快照和日志文件，这个是操作者的责任。
定时清理的方式有很多
a : 自己写脚本
b : ZK自己已经写好了脚本，在bin/zkCleanup.sh中，所以直接使用这个脚本也是可以执行清理工作的。
c: zookeeper在zoo.cfg中提供了两个参数用来做清理工作
参数一：autopurge.purgeInterval 这个参数指定了清理频率，单位是小时，需要填写一个1或更大的整数，默认是0，表示不开启自己清理功能。
参数二：autopurge.snapRetainCount 这个参数和上面的参数搭配使用，这个参数指定了需要保留的文件数目。默认是保留3个。
-----------------------------------------------kafka---------------------------------------------------------
【2 ： kafka集群搭建】
【2.1环境准备】
1、linux一台或多台，大于等于2
2、已经搭建好的zookeeper集群
3、软件版本kafka_2.11-0.9.0.1.tgz
【2.2】解压tar包
在192.168.1.224 hadoop2004机器上
tar -zxvf kafka_2.11-0.9.0.1.tgz /usr/local
【2.3】添加log目录和修改配置
cd /usr/llocal/kafka_2.11-0.9.0.1
mkdir kafka_log
cd /usr/local/kafka_2.11-0.9.0.1/config
vim server.properties
配置文件解释
broker.id=1
#当前机器在集群中的唯一标识，和zookeeper的myid性质一样
port=19092
#当前kafka对外提供服务的端口默认是9092
host.name=192.168.1.224
#这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。
num.network.threads=3
#这个是borker进行网络处理的线程数
num.io.threads=8
#这个是borker进行I/O处理的线程数
log.dirs=/usr/local/kafka/kafka_2.11-0.9.0.1/kafka_log
#消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的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.1.224:2181,192.168.1.225:2181,192.168.1.226:1218
#设置zookeeper的连接端口
实际修改
#broker.id=1 每台服务器的broker.id都不能相同
#hostname
host.name=192.168.1.224
#在log.retention.hours=168 下面新增下面三项
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
#设置zookeeper的连接端口
zookeeper.connect=192.168.1.224:2181,192.168.1.225:2181,192.168.1.226:1218
【2.4启动服务】
#从后台启动Kafka集群（3台都需要启动）
./kafka-server-start.sh -daemon ../config/server.properties
启动kafka.png
【2.5检查服务启动】
#执行命令jps
20348 Jps
4233 QuorumPeerMain
18991 Kafka
【2.6创建一个主题】
#创建Topic（在机器192.168.1.224 hadoop2004机器上）
./kafka-topics.sh --create --zookeeper 192.168.1.224:2181 --replication-factor 2 --partitions 1 --topic lvfang
#解释
--replication-factor 2 #复制两份
--partitions 1 #创建1个分区
--topic #主题为shuaige
创建主题.png
【2.7创建一个发布者】
#创建一个broker，发布者（在机器192.168.1.225 hadoop2005机器上）
./kafka-console-producer.sh --broker-list 192.168.1.225:19092 --topic lvfang
创建一个发布者.png
【2.8创建一个订阅者】（在机器192.168.1.226 hadoop2006机器上）
'''在一台服务器上创建一个订阅者'''
./kafka-console-consumer.sh --zookeeper 192.168.1.226:12181 --topic lvfang --from-beginning
创建好订阅者之后就可以在发布者机器上发布消息，看看订阅者能否接收到
创建一个接收者.png
【2.8其他命令】
#查看主题
./kafka-topics.sh --list --zookeeper localhost:12181
#查看topic状态
/kafka-topics.sh --describe --zookeeper localhost:12181 --topic lvfang
#下面是显示信息
Topic:ssports PartitionCount:1 ReplicationFactor:2 Configs:
Topic: lvfang Partition: 0 Leader: 1 Replicas: 0,1 Isr: 1
#分区为为1 复制因子为2 他的 shuaige的分区为0
#Replicas: 0,1 复制的为0，1
查看主题列表.png
查看主题状态.png
PartitionCount:1 分区数为1个
ReplicationFactor:2 分区备份2个
Topic: lvfang 主题是 lvfang
Partition: 0 在0分区上
Leader: 2 Leader是2号机器（即broker 2上）
Replicas: 2,3 备份在2号机和3号机上
Isr: 2,3 处于同步中（一般如果leader挂掉了，最好选择处于同步的broker机作为新的leader，这样数据误差会最小 ）
OK
kafka集群搭建完毕-------------------------------------------------------------
【3：其他说明】
【3.1 日志说明】
默认kafka的日志是保存在/opt/kafka/kafka_2.10-0.9.0.0/logs目录下的，这里说几个需要注意的日志
server.log
#kafka的运行日志
state-change.log
#kafka他是用zookeeper来保存状态，所以他可能会进行切换，切换的日志就保存在这里
controller.log
#kafka选择一个节点作为“controller”,当发现有节点down掉的时候它负责在游泳分区的所有节点中选择新的leader,这使得Kafka可以批量的高效的管理所有分区节点的主从关系。如果controller down掉了，活着的节点中的一个会备切换为新的controller.
【3.2查看zookeeper上的内容】
./zkCli.sh
#查看目录情况 执行“ls /”
[zk: localhost:2181(CONNECTED) 3] ls /
#显示结果：
[consumers, config, controller, isr_change_notification, admin, brokers, zookeeper, controller_epoch]
上面的显示结果中：只有zookeeper是，zookeeper原生的，其他都是Kafka创建的
#标注一个重要的
[zk: localhost:2181(CONNECTED) 3] get /brokers/ids/1
{"jmx_port":-1,"timestamp":"1456125963355","endpoints["PLAINTEXT://192.168.7.100:19092"],"host":"192.168.7.100","version":2,"port":19092}
#还有一个是查看partion
[zk: localhost:2181(CONNECTED) 4] get /brokers/topics/shuaige/partitions/0
2017-01-13_124134.png
