Kafka 0.9+Zookeeper3.4.6集群搭建配置,新版Java Client的使用要点,高可用性测试,以及各种坑
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Kafka 0.9+Zookeeper3.4.6集群搭建配置,新版Java Client的使用要点,高可用性测试,以及各种坑相关的知识,希望对你有一定的参考价值。
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版本。
1) 首先解压
tar -xzvf zookeeper-3.4.6.tar.gz
2) 进入zookeeper的conf目录,将zoo_sample.cfg复制一份,命名为zoo.cfg,此即为Zookeeper的配置文件
cp zoo_sample.cfg zoo.cfg
3) 编辑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选举使用的端口,需确保三台主机的这两个端口都是互通的
4) 在另外两台主机上执行同样的操作,安装并配置zookeeper
5) 分别在三台主机的dataDir路径下创建一个文件名为myid的文件,文件内容为该zk节点的编号。例如在第一台主机上建立的myid文件内容是0,第二台是1。
6) 启动三台主机上的zookeeper服务:
bin/zkServer.sh start
3个节点都启动完成后,可依次执行如下命令查看集群状态:
bin/zkServer.sh status
命令输出如下:
Mode: leader 或 Mode: follower |
3个节点中,应有1个leader和两个follower
7) 验证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
1) 首先解压
tar -xzvf kafka_2.11-0.9.0.1.tgz
2) 编辑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 能够看到,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地址就可以了。 |
3) 在另一台主机也完成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。 |
4) 创建一个名为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的使用,以及集群高可用性的验证测试。
本文出自 “OCD重症患者” 博客,转载请与作者联系!
以上是关于Kafka 0.9+Zookeeper3.4.6集群搭建配置,新版Java Client的使用要点,高可用性测试,以及各种坑的主要内容,如果未能解决你的问题,请参考以下文章
在使用 Kafka 0.8.1 和 Zookeeper 3.4.6 时遇到 LeaderNotAvailableException