搭建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集群就搭建起来了