进击消息中间件系列（二）：Kafka 单机与集群部署实践

[root@hadoop102 software]# tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/

[hadoop102 config]# vim server.properties

broker.id=0            #broker的id号，集群模式下id号要配置成唯一的
listeners=PLAINTEXT://192.168.81.210:9092   #kafka监听地址
num.network.threads=3             #处理网络请求的线程数量，线程会将接收到的消息放到内存中然后再写入磁盘
num.io.threads=8          #消息从内存写入磁盘时使用的线程数量，主要是用来处理磁盘IO的线程数量
socket.send.buffer.bytes=102400        #发送套接字的缓冲区大小
socket.receive.buffer.bytes=102400       #接收套接字的缓冲区大小
socket.request.max.bytes=104857600       #请求套接字的缓冲区大小
log.dirs=/data/kafka/data/kafka-logs      #消息数据存储路径
num.partitions=1         #topic在当前broker上的分区个数
num.recovery.threads.per.data.dir=1       #用来设置恢复和清理data下数据的线程数量，segment文件默认会被保留7天
offsets.topic.replication.factor=1       #
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168         #segment数据文件保留的期限，单位小时，也就是7天
log.segment.bytes=1073741824       #日志文件中每个segment的大小，默认为1G
log.retention.check.interval.ms=300000      #定期检查segment文件的大小是否超出闲置，单位是毫秒
zookeeper.connect=                         #zookeeper地址
zookeeper.connection.timeout.ms=6000      #zookeeper链接超时时间
group.initial.rebalance.delay.ms=0       #延迟初始消费者重新平衡的时间

scp  -r   $pdir/$fname         $user@$host:$pdir/$fname
命令 递归  要拷贝的文件路径/名称   目的地用户@主机:目的地路径/名称

#在 hadoop102 上，将 hadoop102 中/opt/module/kafka_2.12-3.0.0 目录拷贝到hadoop103上。
[root@hadoop102 ~]# scp -r /opt/module/kafka_2.12-3.0.0/  root@hadoop103:/opt/module/

#也可以在 hadoop104 上，将 hadoop102 中/opt/module/kafka_2.12-3.0.0 目录拷贝到hadoop104上。
[root@hadoop104 ~]# scp -r  root@hadoop102:/opt/module/kafka_2.12-3.0.0   ./

#还可以在 hadoop103 上操作，将 hadoop102 中/opt/modulekafka_2.12-3.0.0 目录拷贝到hadoop104上。
[root@hadoop103 ~]# scp -r  root@hadoop102:/opt/module/kafka_2.12-3.0.0   root@hadoop104:/opt/module

rsync  -av      $pdir/$fname         $user@$host:$pdir/$fname
命令  选项参数  要拷贝的文件路径/名称    目的地用户@主机:目的地路径/名称
-a 归档拷贝
-v 显示复制过程

#删除 hadoop103 中/opt/module/kafka_2.12-3.0.0 
[root@hadoop103 module]# rm -rf kafka_2.12-3.0.0/

#同步 hadoop102 中的/opt/module/kafka_2.12-3.0.0  到 hadoop103
[root@hadoop102 module]# rsync -av /opt/module/kafka_2.12-3.0.0/ root@hadoop103:/opt/module/kafka_2.12-3.0.0/

#rsync 命令原始拷贝：之前hadoop102 同步到hadoop103、hadoop104需要输入以下命令
[root@hadoop102 module]# rsync -av /opt/module/kafka_2.12-3.0.0/ root@hadoop103:/opt/module/kafka_2.12-3.0.0/
[root@hadoop102 module]# rsync -av /opt/module/kafka_2.12-3.0.0/ root@hadoop104:/opt/module/kafka_2.12-3.0.0/

#现在期望只输入以下命令就能同步到所有服务器上
[root@hadoop102 module]# xsync kafka_2.12-3.0.0/

#并且期望脚本xsync kafka_2.12-3.0.0/在任何路径都能使用（所以脚本需要放在声明了全局环境变量的路径下）

1.首先查看哪些目录是全局的
[root@hadoop102 module]# echo $PATH
/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin

2.如果不想在提供的目录下存放脚本, 可以添加一个自己的目录到全局环境中（前提是要创建好/home/hadoop/bin目录）
#在/etc/profile.d/my_env.sh 文件中增加 export PATH="$PATH:/home/hadoop/bin" 环境变量配置
[root@hadoop102 ~]# vim /etc/profile.d/my_env.sh
export PATH="$PATH:/home/hadoop/bin"
#配置完后，刷新一下环境变量
[root@hadoop102 ~]# source /etc/profile

3.再次查看时，多了一个全局目录/home/hadoop/bin
[root@hadoop102 ~]# echo $PATH
/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin:/home/hadoop/bin

#在/home/hadoop/bin 目录下创建 xsync 文件，在该文件中编写如下代码
[root@hadoop102 bin]$ vim xsync

#!/bin/bash
#$#表示参数个数，判断参数个数是否小于1
if [ $# -lt 1 ]
then
 echo Not Enough Arguement!
 exit;
fi
#遍历集群所有机器
for host in hadoop102 hadoop103 hadoop104
do
 #打印输入的主机名
 echo ==================== $host ====================
 #遍历所有目录，逐个发送
 # 例如执行命令：xsync kafka zookeeper，表示先同步kafka文件，再同步zookeeper文件
 for file in $@
 do
 #判断文件是否存在
 if [ -e $file ]
 then
 #获取父目录
 # 先执行cd -P /opt/module，再执行pwd，可以获取当前的父目录，-P表示软链接（P大写）
 pdir=$(cd -P $(dirname $file); pwd)
 #获取当前文件的名称
 # basename /opt/module/kafka， 则当前文件名为kafka
 fname=$(basename $file)
 # ssh hadoop103可以访问hadoop103主机，并且创建目录mkdir -p /opt/module/kafka，-p表示不管目录存不存在都会创建
 ssh $host "mkdir -p $pdir"
 # rsync -av /opt/module/kafka/ root@hadoop103:/opt/module/kafka/
 rsync -av $pdir/$fname $host:$pdir
 else
 echo $file does not exists!
 fi
 done
done

#修改脚本 xsync 具有执行权限
[root@hadoop102 bin]# chmod 777 xsync

#执行脚本
[root@hadoop102 ~]# xsync /opt/module/kafka_2.12-3.0.0/
执行完后，hadoop102、hadoop103、hadoop104上都会有kafka_2.12-3.0.0文件

#ssh 另一台电脑的 IP 地址，例如：hadoop102登录到hadoop103
[root@hadoop102 ~]# ssh hadoop103

#hadoop103 退回到 hadoop102
[atguigu@hadoop103 ~]$ exit

#进入.ssh目录，该目录下有个known_hosts文件，再查看known_hosts文件
[root@hadoop102 ~]# cd .ssh/
[root@hadoop102 .ssh]# ll
total 4
-rw-r--r--. 1 root root 558 Nov 28 00:11 known_hosts
[root@hadoop102 .ssh]# cat known_hosts
hadoop103,192.168.10.103 ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBIW8j1y/nhu5Q4K+3VGjTnChzq9cbIjzozabQQcjUCU9PdVFBIdD8PleMmDBEK6NCHzF7EW1m6n6iA1S4ihC3GM=
hadoop104,192.168.10.104 ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBIW8j1y/nhu5Q4K+3VGjTnChzq9cbIjzozabQQcjUCU9PdVFBIdD8PleMmDBEK6NCHzF7EW1m6n6iA1S4ihC3GM=
hadoop102,192.168.10.102 ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBIW8j1y/nhu5Q4K+3VGjTnChzq9cbIjzozabQQcjUCU9PdVFBIdD8PleMmDBEK6NCHzF7EW1m6n6iA1S4ihC3GM=

说明hadoop102访问过hadoop102、hadoop103、hadoop104

#输入ssh-keygen -t rsa后，一直点击回车键，直到出现以下结果
[root@hadoop102 .ssh]# ssh-keygen -t rsa
Generating public/private rsa key pair.
Enter file in which to save the key (/root/.ssh/id_rsa):
Enter passphrase (empty for no passphrase):
Enter same passphrase again:
Your identification has been saved in /root/.ssh/id_rsa.
Your public key has been saved in /root/.ssh/id_rsa.pub.
The key fingerprint is:
SHA256:NIVyD44Gsv0U5nVOiN5KA79KFCl4sQvlVO7hlQdhFZA root@hadoop102
The key's randomart image is:
+---[RSA 2048]----+
| .+o..=*o+.      |
|.++o=.Eo*.o      |
|..o=oOoO+*       |
| ..+ooXoo.o      |
|  ..o= +S        |
|    . +          |
|   . .           |
|    .            |
|                 |
+----[SHA256]-----+

# 再次查看.ssh目录，此时生成了公钥和私钥id_rsa、id_rsa.pub
[root@hadoop102 .ssh]# ll
total 12
-rw-------. 1 root root 1675 Nov 28 00:54 id_rsa
-rw-r--r--. 1 root root  396 Nov 28 00:54 id_rsa.pub
-rw-r--r--. 1 root root  558 Nov 28 00:11 known_hosts

# 查看公钥
[root@hadoop102 .ssh]# cat id_rsa.pub
ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDC6TzfM8ZEZQ8X0u9zFyWQ65cMuBk8YB3yqEjyEjVYtAHijSg0EpTh3NVrEmWaPaSv+6tCziPYjsdWdIQ1k+h8CgyRnV1vc4UTH+YETcl78ScoLWfDxiEBVmKS4aPMYt21yvExGQ/3wZYGFpGVoiUdpHU/QVbpXPjQZ5KjIkOZxzuqsDx5pUbuLT/TLxB5d0ZGYl5mMmVpFIyPlr+btozGNe23h1KAI/rniIrAqU/jYaeoWcrpFLgmiuoG4DPpPqev+GjBe3CS6ow+2i0UwX9czul+cZMWF58EJrhXX/zD6lC+nr6qX5mUTzuhoQpquYy9YetjhpD4iHOAjwT6sbiX root@hadoop102

# 将hadoop102的公钥复制给hadoop102、hadoop103、hadoop104
[root@hadoop102 .ssh]# ssh-copy-id hadoop102
[root@hadoop102 .ssh]# ssh-copy-id hadoop103
[root@hadoop102 .ssh]# ssh-copy-id hadoop104

# hadoop102再次访问hadoop103时，将不再输入hadoop103的密码
[root@hadoop102 .ssh]# ssh hadoop103
Last login: Mon Nov 28 00:24:06 2022 from hadoop102
[root@hadoop103 ~]#

# 查看hadoop103中的.ssh目录，多了一个authorized_keys文件，该文件尾部有一行root@hadoop102表示允许hadoop102免密登录
[root@hadoop103 .ssh]# ll
total 8
-rw-------. 1 root root 396 Nov 28 00:59 authorized_keys
-rw-r--r--. 1 root root 372 Nov 28 00:51 known_hosts
[root@hadoop103 .ssh]# cat authorized_keys
ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDC6TzfM8ZEZQ8X0u9zFyWQ65cMuBk8YB3yqEjyEjVYtAHijSg0EpTh3NVrEmWaPaSv+6tCziPYjsdWdIQ1k+h8CgyRnV1vc4UTH+YETcl78ScoLWfDxiEBVmKS4aPMYt21yvExGQ/3wZYGFpGVoiUdpHU/QVbpXPjQZ5KjIkOZxzuqsDx5pUbuLT/TLxB5d0ZGYl5mMmVpFIyPlr+btozGNe23h1KAI/rniIrAqU/jYaeoWcrpFLgmiuoG4DPpPqev+GjBe3CS6ow+2i0UwX9czul+cZMWF58EJrhXX/zD6lC+nr6qX5mUTzuhoQpquYy9YetjhpD4iHOAjwT6sbiX root@hadoop102

# 再次进行同步分发给其他服务器，不用再输入密码了
[root@hadoop102 bin]# xsync /opt/module/kafka_2.12-3.0.0/

[root@hadoop103 config]# vim server.properties
修改:broker.id=1
[root@hadoop104 config]# vim server.properties
修改:broker.id=2

#在/etc/profile.d/my_env.sh 文件中增加 kafka 环境变量配置
[root@hadoop102 ~]# vim /etc/profile.d/my_env.sh
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka_2.12-3.0.0
export PATH=$PATH:$KAFKA_HOME/bin

#刷新一下环境变量
[root@hadoop102 ~]# source /etc/profile

#分发环境变量文件到其他服务器，并source /etc/profile
[root@hadoop102 ~]# xsync /etc/profile.d/my_env.sh
[root@hadoop103 ~]$ source /etc/profile
[root@hadoop104 ~]$ source /etc/profile

#先启动 Zookeeper 集群，然后启动 Kafka。
[root@hadoop102 kafka]$ zk.sh start

#依次在 hadoop102、hadoop103、hadoop104 节点上启动 Kafka
[root@hadoop102 kafka_2.12-3.0.0]# bin/kafka-server-start.sh -daemon config/server.properties
[root@hadoop103 kafka_2.12-3.0.0]# bin/kafka-server-start.sh -daemon config/server.properties
[root@hadoop104 kafka_2.12-3.0.0]# bin/kafka-server-start.sh -daemon config/server.properties

[root@hadoop102 kafka_2.12-3.0.0]$ bin/kafka-server-stop.sh 
[root@hadoop103 kafka_2.12-3.0.0]$ bin/kafka-server-stop.sh 
[root@hadoop104 kafka_2.12-3.0.0]$ bin/kafka-server-stop.sh

#在/home/hadoop/bin 目录下创建文件 kf.sh 脚本文件
[root@hadoop102 bin]$ vim kf.sh

脚本如下：

#! /bin/bash
case $1 in
"start"){
 for i in hadoop102 hadoop103 hadoop104
 do
 echo " --------启动 $i Kafka-------"
 ssh $i "/opt/module/kafka_2.12-3.0.0/bin/kafka-server-start.sh -daemon /opt/module/kafka_2.12-3.0.0/config/server.properties"
 done
};;
"stop"){
 for i in hadoop102 hadoop103 hadoop104
 do
 echo " --------停止 $i Kafka-------"
 ssh $i "/opt/module/kafka_2.12-3.0.0/bin/kafka-server-stop.sh "
 done
};;
esac

[root@hadoop102 bin]$ chmod +x kf.sh

[root@hadoop102 ~]$ kf.sh start

[root@hadoop102 ~]$ kf.sh stop