搭建Kafka集群（3-broker）

Apache Kafka是一个分布式消息发布订阅系统，而Kafka环境每每是在集群中配置的。本篇就对配置3个broker的Kafka集群进行介绍。

Zookeeper集群

Kafka自己提供了启动了zookeeper的脚本和配置文件。

1. 修改配置文件

进入kafka主目录，编辑文件config/zookeeper.properties，配置以下：

tickTime=2000 #心跳时间，毫秒
initLimit=5 #Follower在启动是须要在5个心跳时间内从Leader同步数据
syncLimit=2   #超过两个心跳时间收不到Follower的效应，就认为此Follower已经下线 
 
#zookeeper集群的三个服务器
server.1=192.168.10.152:2888:3888
server.2=192.168.10.153:2888:3888
server.3=192.168.10.170:2888:3888

在三个机器进行一样得配置，或者将配置文件传到集群中其余机器上。

2. 配置主机标识

在三台机器dataDir目录(zookeeper.properties配置)下，新建文件myid，写入相应标识

echo 1 > /tmp/zookeeper/myid #1为标识，在其余机器上改成2或3

3. 启动Zookeeper集群

逐渐启动三台机器的zookeeper构成一个集群，即在三台机器上运行

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties #-daemon以守护进程方式启动zookeeper

因为节点Zookeeper启动后，都会尝试链接其余节点，先启动的确定连不上后启动的节点，所以，此时打印的异常能够忽略，当节点所有启动，集群在选出一个Leader后，异常就不会打印了。

检查Zookeeper有没有启动

netstat -apn | grep 2181

搭建Kafka broker集群

1. 修改配置文件

修改config/server.properties文件

broker.id=0  #三个broker的id不能相同，所以改成不一样的id
listeners=PLAINTEXT://192.168.10.152:9092 #服务器监听的地址，若是不配置从java.net.InetAddress.getCanonicalHostName()得到
host.name=192.168.10.152 #broker 机器ip
zookeeper.connect=192.168.10.152:2181,192.168.10.153:2181,192.168.10.170:2181

集群的其余机器也进行相应的配置

2. 启动Kafka服务

在三个机器上启动Kafka服务

bin/kafka-server-start.sh -daemon config/server.properties #-daemon以守护进程方式启动kafka server

检查服务是否启动

ps -aux | grep config/server

测试

1. 建立topic

在任意一台机器上建立topic

bin/kafka-topics.sh --create --zookeeper 192.168.10.152:2181 --replication-factor 3 --partitions 3 --topic cluster-test 
#--zookeeper是集群列表，能够指定全部节点，也能够指定为部分列表
#--replication-factor 为复制数目，数据会自动同步到其余broker上，防止某个broker宕机数据丢失
#--partitions 一个topic能够切分红多个partition，一个消费者能够消费多个partition，但一个partition只能被一个消费者消费

查看topic详情

bin/kafka-topics.sh --describe --zookeeper 192.168.10.153:2181 --topic cluster-test

2. 启动生产者、消费者

启动生产者

bin/kafka-console-producer.sh --broker-list 192.168.10.152:9092 --topic cluster-test

在三台机器上分别启动消费者

bin/kafka-console-consumer.sh --zookeeper 192.168.10.152:2181 --topic cluster-test --from-beginning

3. 测试

在生产者console输入消息进行发送

此时，消费者console能够收到消息了

至此，一个简单的Kafka集群就搭建起来了