kafka分布式部署与验证

kafka是一个分布式环境下的消息组件,如果kafka的应用进程被杀或者kafka的机器宕机,那么kafka消息组件就无法使用了。

kafka集群(cluster)

一台机器不够,那就多搞几台,首先,启动zookeeper这个就不多说了.可以参看前文,在启动kafka的时候,我们在单机模拟启动多个kafka应用.

首先在config目录,copy两个server.properties 文件,这里我复制三份,分别起名server1.properties ,server2.properties server3.properties

然后修改这三个配置文件,主要修改broker.id=2,port=9094,log.dir=/tmp/kafka-logs-2这三个值,broker.id是用来标记分布式环境中的broker的,要求唯一,port和log.dir一个端口,一个log目录,如果在真实的分布式环境中是不需要修改.这里单机模拟防止端口冲突.

分别把broker.id改为1,2,3,log.dir则分别改成kafka-logs-1,kafka-logs-2,kafka-logs-3,然后依次启动

kafka-server-start.bat ../../config/server1.properties

kafka-server-start.bat ../../config/server2.properties

kafka-server-start.bat ../../config/server3.properties

如果你启动有报错,一个就是之前说的那个vm参数太大,另一个可能是你的端口没改好.具体错误看下报错就好了.

然后我们注册一个topic,叫做replicationtest

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic replicationtest

这里冗余是3,分区是1,那么最终各个broker都会保留一份,最多允许N-1,也就是2台broker宕机,服务照样运行.

注册之后,这时候

kafka-topics.bat--describe --zookeeper localhost:2181 --topic replicationtest

执行描述命令,看下集群情况

集群描述结果

第一行给出了分区的汇总信息。每个分区行给出分区信息。

“Leader” 节点是2.

“Replicas” 信息，在节点2,3,1上,所有的节点信息.

“Isr” 工作中的复制节点的集合. 也就是活的节点的集合.

其他的就不用解释了.这里选出了2是leader,也就是说2这个节点会给消费者提供服务.

然后我们测试一条信息.

kafka-console-producer.bat --broker-list localhost:7777,localhost:8888,localhost:9999 --topic replicationtest

上面的7777是server1.properties 中设置的.根据个人情况.改改.然后在控制台发发消息.

然后消费一下.

kafka-console-consumer.bat --zookeeper localhost:2181 --topic replicationtest

这里的2181是zookeeper的端口,不用改.

目前运行结果

然后.我们开始关掉一个broker,在3的控制台里CTRL,C.然后是否终止操作,输入Y.

再发一条消息

broker3宕机

一切正常.我们看一下集群信息

broker3宕机集群

发现Isr中存活的机器少了3.因为3挂了.

然后我们关掉broker2.这时候,会触发新的leader选举.期望值1变成leader,再发一条消息

broker2宕机

可以看到生产者发消息过程中,产生了异常,因为和2的连接断开了.但是注意,消息并没有丢,因为触发了新的选举.可以看到,消费者还是接到了正常的消息.集群情况如下

broker2宕机集群

至此,kafka的broker集群测试完毕,那么剩下的问题来了.消费者启动的时候连接的是zookeeper的地址,如果这台zookeeper挂了呢.

那么我们需要zookeeper集群部署.

zookeeper集群

这就包括两部分.

是broker本来要能知道这些zookeeper集群的地址,当一个宕机的时候,才会切换到另一个zookeeper

消费者要知道这些zookeeper的地址,理由同上.

因此步骤如下.可以自己试一试,比较简单

复制3份zookeeper.properties文件,命名为 zookeeper1.properties,zookeeper2.properties,zookeeper3.properties,修改文件中的 dataDir=/tmp/zookeeper和,clientPort=2181,端口分别设置为2181,2182,2183.然后启动三个 zookeeper

修改kafka启动配置,server1.properties三个文件中的zookeeper.connect=localhost:2181 这个配置,逗号隔开.最终为 zookeeper.connect=localhost:2181,localhost:2182,localhost:2183,然后启动

生产者也改下配置中的.metadata.broker.list=localhost:9092,如果使用命令行启动就不用改了.参数指定也可以.

消费者同理,可以改下配置文件中zookeeper.connect=127.0.0.1:2181,也可以命令行启动的时候修改.

5.最终就是各种宕机测试了.