Kafka——集群安裝部署(自带Zookeeper)

来自CloudWiki
跳转至: 导航搜索

kafka简介

kafka官网:http://kafka.apache.org/

kafka下载页面:http://kafka.apache.org/downloads

kafka配置快速入门:http://kafka.apache.org/quickstart

新版本的kafka自带有zookeeper,本篇文章记录使用自带zookeeper搭建kafka集群。

环境准备

kafka版本:kafka_2.11-2.3.1.tgz

三台主机IP:192.168.128.130(master),192.168.128.131,192.168.128.132

下载解压kafka

在master、slave1、slav2上进行如下操作:

解压kafka

上传kafka安装包到slave1的/opt目录,解压缩

tar -zxf kafka_2.11-2.3.1.tgz -C /usr/local/

查看配置文件

[root@master ~]# cd /usr/local

[root@master local]# ls

apache-hive-1.2.1-bin  hbase-1.3.6       scala-2.11.5
bin                    include           share
etc                    kafka_2.11-2.3.1  spark-2.4.0-bin-hadoop2.6
flink-1.9.2            lib               src
flume                  lib64             zookeeper-3.4.6
games                  libexec
hadoop-2.6.5           sbin

[root@master local]# cd kafka_2.11-2.3.1/

[root@master kafka_2.11-2.3.1]# ll config

total 68
-rw-r--r--. 1 root root  906 Oct 18  2019 connect-console-sink.properties
-rw-r--r--. 1 root root  909 Oct 18  2019 connect-console-source.properties
-rw-r--r--. 1 root root 5321 Oct 18  2019 connect-distributed.properties
-rw-r--r--. 1 root root  883 Oct 18  2019 connect-file-sink.properties
-rw-r--r--. 1 root root  881 Oct 18  2019 connect-file-source.properties
-rw-r--r--. 1 root root 1552 Oct 18  2019 connect-log4j.properties
-rw-r--r--. 1 root root 2262 Oct 18  2019 connect-standalone.properties
-rw-r--r--. 1 root root 1221 Oct 18  2019 consumer.properties
-rw-r--r--. 1 root root 4727 Oct 18  2019 log4j.properties
-rw-r--r--. 1 root root 1925 Oct 18  2019 producer.properties
-rw-r--r--. 1 root root 6851 Oct 18  2019 server.properties
-rw-r--r--. 1 root root 1032 Oct 18  2019 tools-log4j.properties
-rw-r--r--. 1 root root 1169 Oct 18  2019 trogdor.conf
-rw-r--r--. 1 root root 1023 Oct 18  2019 zookeeper.properties

其中server.properties 和 zookeeper.properties 分别对应kafka和zookeeper的配置文件

建立日志目录

[root@master kafka_2.11-2.3.1]# pwd

/usr/local/kafka_2.11-2.3.1

[root@master kafka_2.11-2.3.1]# mkdir zookeeper #创建zookeeper数据目录

[root@master kafka_2.11-2.3.1]# mkdir log #创建日志目录

[root@master kafka_2.11-2.3.1]# mkdir log/zookeeper #创建zookeeper日志目录

[root@master kafka_2.11-2.3.1]# mkdir log/kafka #创建kafka日志目录

zookeeper配置

在master、slave1、slav2上进行如下操作:

修改 zookeeper.properties

进入config目录下,修改关键配置如下,3台服务器的zookeeper.properties配置文件都一样

##修改为自定义的zookeeper数据目录
dataDir=/usr/local/kafka_2.11-2.3.1/zookeeper

#修改为自定义的zookeeper日志目录
dataLogDir=/usr/local/kafka_2.11-2.3.1/log/zookeeper

# 端口
clientPort=2181

#注释掉
#maxClientCnxns=0

#设置连接参数,添加如下配置
tickTime=2000
initLimit=10
syncLimit=5

server.0=192.168.128.130:2888:3888
server.1=192.168.128.131:2888:3888
server.2=192.168.128.132:2888:3888

zookeeper数据目录添加myid配置

在各台服务器的zookeeper数据目录【/usr/local/kafka_2.11-2.3.1/zookeeper】添加myid文件,写入服务broker.id属性值

如这里的目录是/usr/local/kafka_2.11-1.0.0/zookeeper

注:3臺服務器都要添加myid文件,但值的內容不一樣,具體關係請見下圖:

第一台broker.id为0的服务

[root@master kafka_2.11-2.3.1]# cd zookeeper

[root@master zookeeper]# pwd

/usr/local/kafka_2.11-2.3.1/zookeeper

[root@master zookeeper]# echo 0 > myid

[root@master zookeeper]# cat myid

0

第二台broker.id为1的服务

[root@slave1 kafka_2.11-2.3.1]# cd zookeeper

[root@slave1 zookeeper]# pwd

/usr/local/kafka_2.11-2.3.1/zookeeper

[root@slave1 zookeeper]# echo 1 > myid

[root@slave1 zookeeper]# cat myid

1

第三台broker.id为2的服务

[root@slave2 kafka_2.11-2.3.1]# cd zookeeper

[root@slave2 zookeeper]# ls

[root@slave2 zookeeper]# echo 2 > myid

[root@slave2 zookeeper]# cat myid

2

kafka配置

进入/usr/local/kafka_2.11-2.3.1/config/,修改配置文件server.properties

broker.id=0
# broker 的全局唯一编号,不能重复
#advertised.listeners=PLAINTEXT://192.168.128.130:9092 # 配置监听,修改为本机ip
listeners=PLAINTEXT://192.168.128.130:9092
#修改为本机ip

############################# Log Basics #############################
# kafka 运行日志存放路径
log.dirs=/usr/local/kafka_2.11-2.3.1/log/kafka
# topic 在当前broker上的分片个数,与broker保持一致
num.partitions=3


############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
#zookeeper.connect=localhost:2181
# 配置三台服务zookeeper连接地址
zookeeper.connect=192.168.128.130:2181,192.168.128.131:2181,192.168.128.132:2181


scp -r /usr/local/kafka_2.11-2.3.1 slave1:/usr/local/

scp -r /usr/local/kafka_2.11-2.3.1 slave2:/usr/local/

注:执行这两步操作后,上面zookeeper的myid 因为被覆盖可能需要再去改一下

同时,slave1、slave2这两个节点的server.properties中的broker.id 和advertised.listeners也需要改一下

进入/usr/local/kafka_2.11-2.3.1/config/,修改配置文件server.properties

slave1:

broker.id=1 # broker 的全局唯一编号,不能重复
#advertised.listeners=PLAINTEXT://192.168.128.131:9092 # 配置监听,修改为本机ip
listeners=PLAINTEXT://192.168.128.131:9092
#修改为本机ip

slave2:

broker.id=2 # broker 的全局唯一编号,不能重复
#advertised.listeners=PLAINTEXT://192.168.128.132:9092 # 配置监听,修改为本机ip
listeners=PLAINTEXT://192.168.128.132:9092
#修改为本机ip

kafka启动

kafka启动时先启动zookeeper,再启动kafka;关闭时相反,先关闭kafka,再关闭zookeeper

启动zookeeper

在master、slave1、slave2三个节点上输入如下命令:

[root@slave1 kafka_2.11-2.3.1]# pwd

/usr/local/kafka_2.11-2.3.1

[root@slave1 kafka_2.11-2.3.1]# bin/zookeeper-server-start.sh config/zookeeper.properties &


  1. 後臺啟動命令:

[root@localhostkafka_2.11-1.0.0]# nohup bin/zookeeper-server-start.sh config/zookeeper.properties >log/zookeeper/zookeeper.log 2>1 &

启动kafka

  1. 後臺啟動命令:

[root@localhostkafka_2.11-1.0.0]# nohup bin/kafka-server-start.sh config/server.properties >log/kafka/kafka.log 2>1 &

在3个节点输入 查看zookeeper 和kafka是否启动?

[root@slave2 kafka_2.11-2.3.1]# jps

3188 Kafka
3544 Jps
2846 QuorumPeerMain

kafka测试验证

分别启动三台机器的zookeeper,三台机器的zookeeper都启动成功后,再分别启动三台机器的kafka。

進入kafka根目錄

cd /usr/local

cd kafka_2.11-2.3.1/

创建topic

在某臺機器創建topic,名稱為test

[root@master kafka_2.11-2.3.1]# bin/kafka-topics.sh -create --zookeeper 192.168.128.130:2181,192.168.128.131:2181, 192.168.128.132:2181 -replication-factor 3 --partitions 3 --topic test

Created topic test.

命令解析:

--create:          #指定创建topic动作

--topic:          #指定新建topic的名称

--zookeeper:        #指定kafka连接zk的连接url,该值和server.properties文件中的配置项{zookeeper.connect}一样

--partitions 3       #指定当前创建的kafka分区数量,默认为1个

--replication-factor 3  #指定每个分区的副本集数量,不能大于 broker 节点数量,多了也没用,1个节点放>=2个副本挂了都完蛋。

查看topic

查看topic,确认topic创建成功

[root@master kafka_2.11-2.3.1]# bin/kafka-topics.sh --list --zookeeper 192.168.128.130:2181

test

查看topic,详细信息

[root@master kafka_2.11-2.3.1]# bin/kafka-topics.sh --describe --zookeeper 192.168.128.130:2181 --topic test

Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 2       Replicas: 2,0,1 Isr: 2,0,1
        Topic: test     Partition: 1    Leader: 0       Replicas: 0,1,2 Isr: 0,1,2
        Topic: test     Partition: 2    Leader: 1       Replicas: 1,2,0 Isr: 1,2,0

可以描述为:Topic分区数/副本数/副本Leader/副本ISR等信息:

  “leader”:该节点负责该分区的所有的读和写,每个节点的leader都是随机选择的。   “replicas”:备份的节点列表,无论该节点是否是leader或者目前是否还活着,只是显示。   “isr”:同步备份”的节点列表,也就是活着的节点并且正在同步leader      其中Replicas和Isr中的1,2,0就对应着3个broker他们的broker.id属性!

创建生产者

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

broker-list: 本机ip

创建消费者

[root@slave1 kafka_2.11-2.3.1]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.128.131:9092 --topic test --from-beginning

broker-list: 本机ip

由生产者向消费者发信息

生产者:

[root@master kafka_2.11-2.3.1]# bin/kafka-console-producer.sh --broker-list 192.168.128.130:9092 --topic test

>hello 01012
>20200725

消费者:

[root@slave2 kafka_2.11-2.3.1]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.128.132:9092 --topic test --from-beginning

hello 01012
20200725


参考文档:

kafka自带zookeeper: https://www.cnblogs.com/caoweixiong/p/11060533.html