KAFKA最新版 3.0.0集群部署测试 _最新版

大鹏一日同风起，扶摇直上九万里。这篇文章主要讲述KAFKA最新版 3.0.0集群部署测试相关的知识，希望能为你提供帮助。
关于kafka的信息可以参考官方文档： https://kafka.apache.org/documentation/（如果英文不好直接右击翻译成中文就可以，比看网上一堆文章都强）
环境信息：

IP	角色
10.0.83.71	zookeeper、kafka
10.0.83.72	zookeeper、kafka
10.0.83.73	zookeeper、kafka

先把主机名改掉，分别到3台机器上改自己的：

hostnamectl set-hostname kafka1 hostnamectl set-hostname kafka2 hostnamectl set-hostname kafka3### 到3台机器上编辑/etc/hosts文件 10.0.83.71kafka1 10.0.83.72kafka2 10.0.83.73kafka3

分别到3台机器上安装java8

yum -y install java-1.8.0-openjdk-devel

关掉防火墙（如果不关，等下把集群通讯的8182，8183，8184几个开放也可以）

[root@kafka1 ~]# systemctl stop firewalld [root@kafka2 ~]# systemctl stop firewalld [root@kafka3 ~]# systemctl stopfirewalld

https://www.apache.org/dyn/closer.cgi?path=/kafka/3.0.0/kafka_2.12-3.0.0.tgz
下载kafka，并修改配置文件https://dlcdn.apache.org/kafka/3.0.0/kafka_2.12-3.0.0.tgz
tar zxvf kafka_2.12-3.0.0.tgz -C /opt/
cd /opt/kafka_2.12-3.0.0/config/
配置zookeeper集群，修改配置文件zookeeper.properties

vim zookeeper.properties tickTime=2000 initLimit=20 syncLimit=10 server.1=10.0.83.71:2182:2183 server.2=10.0.83.72:2182:2183 server.3=10.0.83.73:2182:2183

tickTime: 这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳
nitLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 5个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒
syncLimit：这个配置项标识 Leader 与Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是5*2000=10秒
clientPort：这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求
server.1 这个1是服务器的标识也可以是其他的数字，表示这个是第几号服务器，用来标识服务器，这个标识要写到快照目录下面myid文件里
10.0.83.71为集群里的IP地址，第一个端口是master和slave之间的通信端口，默认是2182，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是2183）
创建zookeeper所需的目录
分别在3个机器上执行：第一台：

[root@kafka1 ~]# mkdir -p /data/zookeeper [root@kafka1 ~]# echo "1" > /data/zookeeper/myid

第二台：

[root@kafka2 ~]# mkdir -p /data/zookeeper [root@kafka2 ~]# echo "2" > /data/zookeeper/myid

第三台：

[root@kafka3 ~]# mkdir -p /data/zookeeper [root@kafka3 ~]# echo "3" > /data/zookeeper/myid

启动Zookeeper

[root@kafka1 ~]# cd /opt/kafka_2.12-3.0.0/ [root@kafka1 kafka_2.12-3.0.0]# ./bin/zookeeper-server-start.sh./config/zookeeper.properties[root@kafka2 ~]# cd /opt/kafka_2.12-3.0.0/ [root@kafka2 kafka_2.12-3.0.0]# ./bin/zookeeper-server-start.sh./config/zookeeper.properties[root@kafka3 ~]# cd /opt/kafka_2.12-3.0.0/ [root@kafka3 kafka_2.12-3.0.0]# ./bin/zookeeper-server-start.sh./config/zookeeper.properties

效果图：

文章图片

配置Kafka集群环境 1）修改Kafka配置文件config/server.properties

注意注释掉所有节点的broker.id [root@kafka2 ~]# cd /opt/kafka_2.12-3.0.0/ [root@kafka2 kafka_2.12-3.0.0]# vim config/server.properties log.dirs=/data/kafka-logs #broker.id=0 zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181 listeners=PLAINTEXT://10.0.83.72:9092 advertised.listeners=PLAINTEXT://10.0.83.72:9092

创建日志目录：

mkdir -p /data/kafka-logs

分别在3台机器上启动kafka

cd /opt/kafka_2.12-3.0.0 ./bin/kafka-server-start.sh./config/server.properties

启动之后如果不放心可以检查下zookeeper的2181和kafka的9092端口是否正常

[root@kafka3 kafka_2.12-3.0.0]# netstat -nlpt | grep -E "9092|2181" tcp600 10.0.83.73:9092:::*LISTEN147669/java tcp600 :::2181:::*LISTEN146348/java[root@kafka1 kafka_2.12-3.0.0]# netstat -nlpt | grep -E "9092|2181" tcp600 10.0.83.71:9092:::*LISTEN138011/java tcp600 :::2181:::*LISTEN136719/java[root@kafka2 kafka_2.12-3.0.0]# netstat -nlpt | grep -E "9092|2181" tcp600 10.0.83.72:9092:::*LISTEN132679/java tcp600 :::2181:::*LISTEN131438/java

创建topic测试：

[root@kafka1 kafka_2.12-3.0.0]# ./bin/kafka-topics.sh --create --bootstrap-server kafka1:9092 --replication-factor 1 --partitions 1 --topic tp1 Created topic tp1.

查看所有的topic

[root@kafka1 kafka_2.12-3.0.0]# ./bin/kafka-topics.sh --list--bootstrap-server kafka1:9092 tp1

模拟一个数据

[root@kafka1 kafka_2.12-3.0.0]# ./bin/kafka-console-producer.sh --bootstrap-server kafka1:9092 --topic tp1 > 测试数据 > ceshi1

【KAFKA最新版 3.0.0集群部署测试】起一个消费者：

[root@kafka1 kafka_2.12-3.0.0]# ./bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic tp1 测试数据 ceshi1

到这里kafka集群已经ok 把zookeeper和kafka做成系统服务并开机自动启动（可以把之前在终端执行的启动命令关了，不然起不来）在3台机器上分别运行

[root@kafka1 ~]# mkdir /etc/cluster/

创建zookeeper的service文件

[root@kafka1 cluster]# cat /etc/cluster/zookeeper.service [Unit] Description=zookeeper[Service] ExecStart=/opt/kafka_2.12-3.0.0/bin/zookeeper-server-start.sh /opt/kafka_2.12-3.0.0/config/zookeeper.properties SyslogIdentifier=zookeeper[Install] WantedBy=multi-user.target

加为系统服务并开机启动

[root@kafka1 cluster]# ln -s /etc/cluster/zookeeper.service /lib/systemd/system[root@kafka1 cluster]# systemctl start zookeeper### 开机启动： [root@kafka1 cluster]# systemctl enable zookeeper

设置kafka系统服务并开机启动

[root@kafka1 cluster]# cat /etc/cluster/kafka.service [Unit] Description=kafka[Service] ExecStart=/opt/kafka_2.12-3.0.0/bin/kafka-server-start.sh /opt/kafka_2.12-3.0.0/config/server.properties SyslogIdentifier=kafka[Install] WantedBy=multi-user.target

设置开机启动

[root@kafka1 cluster]# ln -s /etc/cluster/kafka.service /lib/systemd/system/ [root@kafka1 cluster]# systemctl start kafka [root@kafka1 cluster]# systemctl enable kafka

查看服务状态：

[root@kafka1 cluster]# systemctl status kafka ● kafka.service - kafka Loaded: loaded (/etc/cluster/kafka.service; enabled; vendor preset: disabled) Active: active (running) since Sun 2021-12-05 20:49:52 EST; 45min ago Main PID: 138011 (java) Tasks: 74 (limit: 48706) Memory: 383.6M

查看服务日志：