KAFKA最新版 3.0.0集群部署测试

Posted 2021-12-07 马鹏飞

关于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.71  kafka1
10.0.83.72  kafka2
10.0.83.73  kafka3

分别到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 stop  firewalld

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"
tcp6       0      0 10.0.83.73:9092         :::*                    LISTEN      147669/java         
tcp6       0      0 :::2181                 :::*                    LISTEN      146348/java  

[root@kafka1 kafka_2.12-3.0.0]# netstat -nlpt | grep -E "9092|2181"
tcp6       0      0 10.0.83.71:9092         :::*                    LISTEN      138011/java         
tcp6       0      0 :::2181                 :::*                    LISTEN      136719/java

[root@kafka2 kafka_2.12-3.0.0]# netstat -nlpt | grep -E "9092|2181"
tcp6       0      0 10.0.83.72:9092         :::*                    LISTEN      132679/java         
tcp6       0      0 :::2181                 :::*                    LISTEN      131438/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

起一个消费者：

[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

查看服务日志：

[root@kafka1 cluster]# journalctl -u kafka -f