为了演示集群的效果，这里准备一台虚拟机(window 7)，在虚拟机中搭建了单IP多节点的zookeeper集群（多IP节点的也是同理的），并且在本机（win 7）和虚拟机中都安装了kafka。
前期准备说明：
1.三台zookeeper服务器，本机安装一个作为server1，虚拟机安装两个(单IP)
2.三台kafka服务器，本机安装一个作为server1，虚拟机安装两个。
备注：当然你可以直接在虚拟机上安装三个服务器分别为server1、server2、server3 。
- 虚拟机和本机网络环境
1.将虚拟机的网络模式调整为桥接模式，将虚拟机的防火墙功能关闭；
2.将主机的防火墙也关闭。
3.互相ping，测试是否能正常连接.。
- 下载kafka
从Kafka官网http://kafka.apache.org/downloads下载Kafka安装包。(要下载Binary downloads这个类型，不要下载源文件，方便使用)
- 下载zookeeper
从zookeeper官网http://zookeeper.apache.org/releases.html下载zookeeper安装包。
- 下载JDK
从官网http://www.java.com/download/下载(建议下载Oracle官方发布的Java),，配置环境变量。
zookeeper集群环境搭建
本机配置
修改本机sever1的zoo.cfg文件配置:
#存储内存中数据库快照的位置，如果不设置参数，更新事务日志将被存储到默认位置。 #dataDir=/tmp/zookeeper dataDir=D:/bigData/zookeeper-3.4.10/data #日志文件的位置 dataLogDir=D:/bigData/zookeeper-3.4.10/zlog #监听端口 clientPort=2181 #集群服务器配置 server.1=192.168.1.130:9000:7000 server.2=192.168.1.103:9001:7001 server.3=192.168.1.103:9002:7002
格式： server.A = B:C:D
A:是一个数字，表示第几号服务器
B:服务器IP地址
C:是一个端口号，用来集群成员的信息交换,表示这个服务器与集群中的leader服务器交换信息的端口
D:是在leader挂掉时专门用来进行选举leader所用的端口
完整配置文件如下：
# The number of milliseconds of each tick tickTime=2000 # The number of ticks that the initial # synchronization phase can take initLimit=10 # The number of ticks that can pass between # sending a request and getting an acknowledgement syncLimit=5 # the directory where the snapshot is stored. # do not use /tmp for storage, /tmp here is just # example sakes. #dataDir=/tmp/zookeeper dataDir=D:/bigData/zookeeper-3.4.10/data dataLogDir=D:/bigData/zookeeper-3.4.10/zlog # the port at which the clients will connect clientPort=2181 # the maximum number of client connections. # increase this if you need to handle more clients #maxClientCnxns=60 # # Be sure to read the maintenance section of the # administrator guide before turning on autopurge. # # http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance # # The number of snapshots to retain in dataDir #autopurge.snapRetainCount=3 # Purge task interval in hours # Set to "0" to disable auto purge feature #autopurge.purgeInterval=1 server.1=192.168.1.130:9000:7000 server.2=192.168.1.101:9001:7001 server.3=192.168.1.101:9002:7002
创建serverID
在zoo.cfg配置文件中dataDir目录中新建一个没有后缀的myid文件，里边写1。
虚拟机配置
将本机的zookeeper安装包拷贝到虚拟机上，然后在盘中新建两个目录，server2,server3目录。
修改server2的zoo.cfg配置文件：
#存储内存中数据库快照的位置，如果不设置参数，更新事务日志将被存储到默认位置。 #dataDir=/tmp/zookeeper dataDir=E:/bigData/server2/zookeeper-3.4.10/data #日志文件的位置 dataLogDir=E:/bigData/server2/zookeeper-3.4.10/zlog #监听端口 clientPort=2182 #集群服务器配置 server.1=192.168.1.130:9000:7000 server.2=192.168.1.103:9001:7001 server.3=192.168.1.103:9002:7002
创建SeverID
在zoo.cfg配置文件中dataDir目录中新建一个没有后缀的myid文件，里边写2。
同理配置server3中的zookeeper，这样zookeeper集群的配置工作已经完成。
下面我们启动zookeeper集群：一个一个启动每一个zookeeper服务器：
先后进入每一个zookeeper服务器的bin目录下执行zkserver.cmd,在先启动的zookeeper会打印其他zookeeper服务器未启动的错误信息，直至到最后一个启动的时候就没有异常信息了。正常界面：
zookeeper集群环境搭建遇到问题总结:
- zookeeper启动时报Cannot open channel to X at election address Error contacting service. It is probably not running.
解决方法：
a. 关闭所有服务器防火墙
b. 将本服务器的IP修改成0.0.0.0
例如：server.1=0.0.0.0:9000:7000
Kafka集群环境搭建：
下载kafka后解压到某个目录，然后修改server.properties配置。
server.properties修改说明：
broker.id=0 • 在kafka这个集群中的唯一标识，且只能是正整数 port=9091 • 该服务监听的端口 host.name=192.168.1.130 • broker 绑定的主机名称(IP) 如果不设置将绑定所有的接口。 advertised.host.name=192.168.1.130 • broker服务将通知消费者和生产者 换言之，就是消费者和生产者就是通过这个主机(IP)来进行通信的。如果没有设置就默认采用host.name。 num.network.threads=2 • broker处理消息的最大线程数，一般情况是CPU的核数 num.io.threads=8 • broker处理IO的线程数 一般是num.network.threads的两倍 socket.send.buffer.bytes=1048576 • socket发送的缓冲区。socket调优参数SO_SNDBUFF socket.receive.buffer.bytes=1048576 • socket接收的缓冲区 socket的调优参数SO_RCVBUF socket.request.max.bytes=104857600 • socket请求的最大数量，防止serverOOM。 log.dirs=logs • kafka数据的存放地址，多个地址的话用逗号隔开。多个目录分布在不同的磁盘上可以提高读写性能 num.partitions=2 • 每个tipic的默认分区个数，在创建topic时可以重新制定 log.retention.hours=168 • 数据文件的保留时间 log.retention.minutes也是一个道理。 log.segment.bytes=536870912 • topic中的最大文件的大小 -1表示没有文件大小限制 log.segment.bytes 和log.retention.minutes 任意一个 达到要求 都会删除该文件 在创建topic时可以重新制定。若没有.则选取该默认值 log.retention.check.interval.ms=60000 • 文件大小检查的周期时间，是否处罚 log.cleanup.policy中设置的策略 log.cleaner.enable=false • 是否开启日志清理 zookeeper.connect=192.168.1.130:num1,192.168.1.130:num2,192.168.1.130:num3 • 上面我们的Zookeeper集群 zookeeper.connection.timeout.ms=1000000 • 进群链接时间超时
本机配置
修改本机server1的server.properties配置：
#在kafka这个集群中的唯一标识，且只能是正整数 broker.id=0 # kafka集群的地址 broker.list=192.168.0.130:9092,192.168.0.103:9093,192.168.0.103:9094 #listener and port port=9092 #broker 绑定的主机名称(IP) 如果不设置将绑定所有的接口。 host.name=192.168.1.130 #kafka数据的存放地址，多个地址的话用逗号隔开。多个目录分布在不同的磁盘上可以提高读写性能 # A comma separated list of directories under which to store log files #log.dirs=/tmp/kafka-logs log.dirs=D:/bigData/kafka_2.11-1.1.0/kafka-logs #每个tipic的默认分区个数，在创建topic时可以重新制定 num.partitions=3 # root directory for all kafka znodes. zookeeper.connect=192.168.1.130:2181,192.168.1.103:2182,192.168.1.103:2183
虚拟机配置
将本机的kafka安装包拷贝到虚拟机上，存放在server2,server3目录。
注意：若本机拷贝到虚拟机中kafka之前有使用过，生成过topics，直接拷贝搭建集群，启动kafka时会报如下错误：
原因：在单点环境下创建的topic中，kafka服务器的broker.id为0，在搭建集群时修改了borker.id，先前创建的topic就无法找到对应的broker.id。
解决办法：删除原来的topic信息。
修改server2目录中kafka配置：
#在kafka这个集群中的唯一标识，且只能是正整数 broker.id=1 # kafka集群的地址 broker.list=192.168.0.130:9092,192.168.0.103:9093,192.168.0.103:9094 #listener and port port=9093 #broker 绑定的主机名称(IP) 如果不设置将绑定所有的接口。 host.name=192.168.1.103 #kafka数据的存放地址，多个地址的话用逗号隔开。多个目录分布在不同的磁盘上可以提高读写性能 # A comma separated list of directories under which to store log files #log.dirs=/tmp/kafka-logs log.dirs=D:/bigData/server2/kafka_2.11-1.1.0/kafka-logs #每个tipic的默认分区个数，在创建topic时可以重新制定 num.partitions=3 # root directory for all kafka znodes. zookeeper.connect=192.168.1.130:2181,192.168.1.103:2182,192.168.1.103:2183
同理修改server3中的kafka配置，配置好之后，kafka集成环境已经配置完成，下面我们来验证下集成环境是否OK。
验证集群环境
依次先启动zookeeper服务器
进入zookeeper的bin目录，执行zkServer.cmd
依次启动kafka服务器
进入kafka安装目录D:igDatakafka_2.11-1.1.0,按下shift+鼠标右键，选择"在此处打开命令窗口"，打开命令行,在命令行中输入：.inwindowskafka-server-start.bat .configserver.properties回车。正常启动界面如图：
在主机kafka服务器创建topic
主要参数说明：
partitions分区数
(1). partitions ：分区数，控制topic将分片成多少个log。可以显示指定，如果不指定则会使用broker(server.properties)中的num.partitions配置的数量
(2). 虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多，会增加不可用及延迟的风险。因为多的分区数，意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗。
(3). 分区数也限制了consumer的并行度，即限制了并行consumer消息的线程数不能大于分区数
(4). 分区数也限制了producer发送消息是指定的分区。如创建topic时分区设置为1，producer发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的；这种情况可以通过alter –partitions 来增加分区数。
replication-factor副本
1. replication factor 控制消息保存在几个broker(服务器)上，一般情况下等于broker的个数。
2. 如果没有在创建时显示指定或通过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量。
创建Topic
1. 创建主题，命名为"test20180430"，replicationfactor=3(因为这里有三个kafka服务器在运行)。可根据集群中kafka服务器个数来修改replicationfactor的数量，以便提高系统容错性等。
2. 在D:igDatakafka_2.11-1.1.0inwindows目录下打开新的命令行，输入命令：kafka-topics.bat --create --zookeeper 192.168.1.130:2181 --replication-factor 3 --partitions 1 --topic test20180430回车。如图：
查看Topic
kafka-topics.bat --zookeeper 192.168.1.130:2181 --describe --topic test20180430
结果说明:
第一行，列出了topic的名称，分区数(PartitionCount),副本数(ReplicationFactor)以及其他的配置(Config.s)
Leader:1 表示为做为读写的broker的编号
Replicas:表示该topic的每个分区在那些borker中保存
Isr:表示当前有效的broker, Isr是Replicas的子集
创建Producer消息提供者
1. 在D:igDatakafka_2.11-1.1.0inwindows目录下打开新的命令行，输入命令：kafka-console-producer.bat --broker-list 192.168.1.130:9092,192.168.1.101:9093,192.168.1.101:9094 --topic test20180430 回车。(该窗口不要关闭)
创建Customer消息消费者
1. 在D:igDatakafka_2.11-1.1.0inwindows目录下打开新的命令行，输入命令：kafka-console-consumer.bat --zookeeper 192.168.1.130:2181 --topic test20180430回车。
以同样的方式，将虚拟机上server2和server3的消息者customer创建，然后通过主机上的producer进行消息发送操作。
至此，集群环境已经验证OK。
集群容错性
1. 首先查看topic的信息
kafka-topics.bat --list --zookeeper 192.168.1.130:2181
查看指定的topic的详细信息：
kafka-topics.bat --zookeeper 192.168.1.130:2181 --describe --topic test20180430
可以看到此时选举的leader是0，即就是虚拟机中的kafka服务器，现在把虚拟机的kafka服务器给干掉。此时leader为变为1，消费者能继续消费。
注意：zk的部署个数最好为基数，ZK集群的机制是只要超过半数的节点OK，集群就能正常提供服务。只有ZK节点挂得太多，只剩一半或不到一半节点能工作，集群才失效。