Kafka 0.9+Zookeeper3.4.6集群搭建、配置，新版Java Client的使用要点，高可用性测试，以及各种坑（一）

kelgon

浏览: 28852 次
性别:
来自: 北京

最近访客更多访客>>

andrew7676

codeeeeeeer

spaceandroid

xchao

博主相关

博客

微博

相册

留言

关于我

文章分类

全部博客 (5)

社区版块

存档分类

kafka 高可用 java 消息 zookeeper

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

# 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.
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，集群中每个Broker的ID不可相同
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

此处的坑：

按照官方文档的说法，advertised.host.name和advertised.port这两个参数用于定义集群向Producer和Consumer广播的节点host和port，如果不定义的话，会默认使用host.name和port的定义。但在实际应用中，我发现如果不定义advertised.host.name参数，使用Java客户端从远端连接集群时，会发生连接超时，抛出异常：org.apache.kafka.common.errors.TimeoutException: Batch Expired

经过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

此处的坑：

官方给出的后台启动kafka的方法是：

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 PartitionCount:2 ReplicationFactor:2 Configs:
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的使用，以及集群高可用性的验证测试。

至下一节的传送门