CentOS7中新版本自带zookeeper搭建Kafka集群

一、zookeeper在kafka中的具体作用

它是一个分布式协调框架。很好的将消息生产、消息存储、消息消费的过程结合在一起。在典型的Kafka集群中，Kafka通过Zookeeper管理集群控制，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费信息。

Kafka严重依赖于Zookeeper集群。所有的broker在启动的时候都会往zookeeper进行注册，目的就是选举出一个controller，controlller会读取注册上的从点的数据（通过监听机制），生成集群的元数据信息，之后把这些信息都分发给其他的服务器，让其他服务器能感知到集群中其他成员的存在。（zookeeper其实主要的存在意义就是给集群的管理机制上面，做保证。都注册到zk，这样就可以相互发现了）

二、集群搭建

1.将kafka_2.12-2.4.1.tgz解压到/export/servers目录下

解压命令：tar -ZXVF kafka_2.12-2.4.1.tgz -C /export/servers

#到/export/servers目录下进行改名
[root@master servers]# mv kafka_2.12-2.4.1.tgz kafka

2、配置Kafka的环境变量

进入到/etc/profile文件中添加以下内容：

#KAFKA_HOME
export KAFKA_HOME=/export/servers/kafka
export PATH=$PATH:${KAFKA_HOME}/bin

3、（1）进入到Kafka的config目录中，修改配置文件server.properties

[root@master config]# vi server.properties 
#修改内容如下
#保证broker唯一，第一台可以不修改默认为0，后面两台需要修改，如改为1和2
broker.id=0

#设置listeners
listeners = PLAINTEXT://master:9092

#kafka的日志目录，这项默认放在tmp目录下，而tmp目录中内容会随着重启而丢失
log.dirs=/export/data/kafka

#分区数量一般与broker保持一致
num.partitions=3

num.recovery.threads.per.data.dir=1

#设置zookeeper.connect，配置三台zookeeper连接地址
zookeeper.connect=master:2181,slave1:2181,slave2:2181

（2）进入到Kafka的config目录中，修改配置文件zookeeper.properties

[root@master config]# vi zookeeper.properties
#配置内容如下
#zookeeper数据目录，这项默认放在tmp目录下，而tmp目录中内容会随着重启而丢失
dataDir=/export/servers/zookeeper/data
#注：mkdir -p /export/servers/zookeeper/data（这是根据自己路径创建的）

#配置连接参数，添加以下配置
sinkTime=2000
initLimit=10
syncLimit=5

#设置broker Id的服务地址
server.0=master:2888:3888
server.1=slave1:2888:3888
server.2=slave2:2888:3888

(3)zookeeper数据目录中添加id配置

[root@master config]# cd /export/servers/zookeeper/data/
[root@master data]# vi myid
#写入broker.id属性值,master中为0，slave1中为1，slave2中为2（每台机子都要创建）
0

3、分发Kafka

scp -r /export/servers/kafka slave1:/export/servers
scp -r /export/servers/kafka slave2:/export/servers

4、修改slave1，slave2中的文件配置

[root@slave1 ~]# cd /export/servers/kafka/
[root@slave1 kafka]# cd config/
[root@slave1 config]# vi server.properties

broker.id=1
listeners = PLAINTEXT://slave1:9092
#随后保存退出（:wq）

[root@slave2 ~]# cd /export/servers/kafka/
[root@slave2 kafka]# cd config/
[root@slave2 config]# vi server.properties

broker.id=2
listeners = PLAINTEXT://slave2:9092
#随后保存退出（:wq）

#slave1，slave2每台机子中myid分别为1，2

三、启动Kafka

kafka启动时先启动zookeeper，在启动kafka；关闭时相反，先关闭kafka，再关闭zookeeper

[root@master data]# cd /export/servers/kafka/bin
[root@master kafka]# bin/zookeeper-server-start.sh config/zookeeper.properties &
#注：三台机子都要启动zookeeper
[root@master kafka]# bin/kafka-server-start.sh config/server.properties &
#启动kafka

四、用jps查看进程，如图：

至此，Kafka集群搭建完毕