Kafka 0.9版本对java client的api做出了较大调整，本文主要总结了Kafka 0.9在集群搭建、高可用性、新API方面的相关过程和细节，以及本人在安装调试过程中踩出的各种坑。
关于Kafka的结构、功能、特点、适用场景等，网上到处都是，我就不再赘述了，直接进入正文
Kafka 0.9集群安装配置
操作系统：CentOS 6.5
1. 安装Java环境
Zookeeper和Kafka的运行都需要Java环境，所以先安装JRE，Kafka默认使用G1垃圾回收器，如果不更改垃圾回收器，官方推荐使用7u51以上版本的JRE。如果你使用老版本的JRE，需要更改Kafka的启动脚本，指定G1以外的垃圾回收器。
Java环境的安装过程在此不赘述了。
2. Zookeeper集群搭建
Kafka依赖Zookeeper管理自身集群（Broker、Offset、Producer、Consumer等），所以先要安装Zookeeper。自然，为了达到高可用的目的，Zookeeper自身也不能是单点，接下来就介绍如何搭建一个最小的Zookeeper集群（3个zk节点）
此处选用Zookeeper的版本是3.4.6，此为Kafka0.9中推荐的Zookeeper版本。
首先解压
tar -xzvf zookeeper-3.4.6.tar.gz
进入zookeeper的conf目录，将zoo_sample.cfg复制一份，命名为zoo.cfg，此即为Zookeeper的配置文件
cp zoo_sample.cfg zoo.cfg
编辑zoo.cfg
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.
dataDir=/data/zk/zk0/data
dataLogDir=/data/zk/zk0/logs
# the port at which the clients will connect
clientPort=2181
server.0=10.0.0.100:4001:4002
server.1=10.0.0.101:4001:4002
server.2=10.0.0.102:4001:4002
- dataDir和dataLogDir的路径需要在启动前创建好
- clientPort为zookeeper的服务端口
- server.0/1/2为zk集群中三个node的信息，定义格式为hostname:port1:port2，其中port1是node间通信使用的端口，port2是node选举使用的端口，需确保三台主机的这两个端口都是互通的
在另外两台主机上执行同样的操作，安装并配置zookeeper
分别在三台主机的dataDir路径下创建一个文件名为myid的文件，文件内容为该zk节点的编号。例如在第一台主机上建立的myid文件内容是0，第二台是1。
接下来，启动三台主机上的zookeeper服务：
bin/zkServer.sh start
3个节点都启动完成后，可依次执行如下命令查看集群状态：
bin/zkServer.sh status
命令输出如下：
Mode: leader 或 Mode: follower
3个节点中，应有1个leader和两个follower
验证zookeeper集群高可用性：
假设目前3个zk节点中，server0为leader，server1和server2为follower
我们停掉server0上的zookeeper服务：
bin/zkServer.sh stop
再到server1和server2上查看集群状态，会发现此时server1（也有可能是server2）为leader，另一个为follower。
再次启动server0的zookeeper服务，运行zkServer.sh status检查，发现新启动的server0也为follower
至此，zookeeper集群的安装和高可用性验证完成。
附：Zookeeper默认会将控制台信息输出到启动路径下的zookeeper.out中，显然在生产环境中我们不能允许Zookeeper这样做，通过如下方法，可以让Zookeeper输出按尺寸切分的日志文件：
修改conf/log4j.properties文件，将
zookeeper.root.logger=INFO, CONSOLE
改为
zookeeper.root.logger=INFO, ROLLINGFILE
修改bin/zkEnv.sh文件，将
ZOO_LOG4J_PROP="INFO,CONSOLE"
改为
ZOO_LOG4J_PROP="INFO,ROLLINGFILE"
然后重启zookeeper，就ok了
3. Kafka集群搭建
此例中，我们会安装配置一个有两个Broker组成的Kafka集群，并在其上创建一个两个分区的Topic
本例中使用Kafka最新版本0.9.0.1
首先解压
tar -xzvf kafka_2.11-0.9.0.1.tgz
编辑config/server.properties文件，下面列出关键的参数
broker.id=0
#监听器，端口号与port一致即可
listeners=PLAINTEXT://:9092
#Broker监听的端口
port=9092
#Broker的Hostname，填主机IP即可
host.name=10.0.0.100
#向Producer和Consumer建议连接的Hostname和port （此处有坑，具体见后）
advertised.host.name=10.0.0.100
advertised.port=9092
#进行IO的线程数，应大于主机磁盘数
num.io.threads=8
#消息文件存储的路径
log.dirs=/data/kafka-logs
#消息文件清理周期，即清理x小时前的消息记录
log.retention.hours=168
#每个Topic默认的分区数，一般在创建Topic时都会指定分区数，所以这个配成1就行了
num.partitions=1
#Zookeeper连接串，此处填写上一节中安装的三个zk节点的ip和端口即可
zookeeper.connect=10.0.0.100:2181,10.0.0.101:2181,10.0.0.102:2181
配置项的详细说明请见官方文档：http://kafka.apache.org/documentation.html#brokerconfigs
此处的坑：
经过debug发现，连接到集群是成功的，但连接到集群后更新回来的集群meta信息却是错误的：
能够看到，metadata中的Cluster信息，节点的hostname是iZ25wuzqk91Z这样的一串数字，而不是实际的ip地址10.0.0.100和101。iZ25wuzqk91Z其实是远端主机的hostname，这说明在没有配置advertised.host.name的情况下，Kafka并没有像官方文档宣称的那样改为广播我们配置的host.name，而是广播了主机配置的hostname。远端的客户端并没有配置hosts，所以自然是连接不上这个hostname的。要解决这一问题，把host.name和advertised.host.name都配置成绝对的ip地址就可以了。
接下来，我们在另一台主机也完成Kafka的安装和配置，然后在两台主机上分别启动Kafka：
bin/kafka-server-start.sh -daemon config/server.properties
此处的坑：
bin/kafka-server-start.sh config/server.properties &
但用这种方式启动后，只要断开Shell或登出，Kafka服务就会自动shutdown，不知是OS的问题还是SSH的问题还是Kafka自己的问题，总之我改用-daemon方式启动Kafka才不会在断开shell后自动shutdown。
接下来，我们创建一个名为test，拥有两个分区，两个副本的Topic：
bin/kafka-topics.sh --create --zookeeper 10.0.0.100:2181,10.0.0.101:2181,10.0.0.102:2181 --replication-factor 2 --partitions 2 --topic test
创建完成后，使用如下命令查看Topic状态：
bin/kafka-topics.sh --describe --zookeeper 10.0.0.100:2181,10.0.0.101:2181,10.0.0.102:2181 --topic test
输出：
Topic: test Partition: 0 Leader: 1 Replicas: 1,0 Isr: 0,1
Topic: test Partition: 1 Leader: 0 Replicas: 0,1 Isr: 0,1
解读：test这个topic，当前有2个分区，分别为0和1，分区0的Leader是1（这个1是broker.id），分区0有两个Replica（副本），分别是1和0，这两个副本中，Isr（In-sync）的是0和1。分区2的Leader是0，也有两个Replica，同样也是两个replica都是in-sync状态
至此，Kafka 0.9集群的搭建工作就完成了，下一节中，我们将介绍新的Java API的使用，以及集群高可用性的验证测试。
