Kafka
简介
传统定义：Kafka是一个分布式的基于发布/订阅模式的消息队列，应用于大数据实时处理领域。
Kafka最新定义：Apache Kafka是一个开源分布式事件流平台，被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用程序
Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications
消息队列的应用场景：
- 缓冲/消峰：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。比如双十一用户发送访问的请求速度和系统处理的速度也是不一样的，就需要消息队列作为缓冲。
- 解耦：允许独立的扩展或修改两边的处理过程，只要保证他们遵守同样的接口约束。
- 异步通信：允许用户把一个消息放入队列，但是不立即处理它，在需要的时候再进行处理。
消息队列的两种模式：
- 点对点模式：一个生产者对应一个消费者，进行点对点的传输消息，消费者主动拉取数据，消息消费之后消除消息。
- 发布/订阅模式：相当于可以分成多个不同的主题的消息，消费者只会接受自己想要的主题内的消息，消费者消费数据后不删除数据，每个消费者之间相互独立，都可以消费到消息。
基本结构
Kafka由以下几个部分构成
- 生产者：向kafka发送消息的客户端称作生产者
- 消费者：向Kafka获取消息的客户端称作消费者
- Broker：每一台Kafka服务器称之为Broker，一个集群由多个Broker组成
- Topic：主题，类似于存储消息的队列，生产者和消费者根据对应主题进行发送、接收消息
- Partition：分区，每一个Topic主题可以对应多个分区，也就是将一个大的Topic拆分成几个小的partiton，可以存储到不同的Broker上
- Replica：副本，为了让集群中节点发送故障时，该节点上的分区数据不丢失，每一个Topic的每个分区都有副本，副本之间存在一个Leader（生产者发送消息、消费者消费消息都是通过Leader）和若干Follower（实时从Leader同步数据，保存和Leader数据的同步）
- Controller：Controller可以看作Broker的领导者，用于管理和协调Kafka集群，比如管理集群中所有分区的状态，当副本的Leader所在节点崩溃，Controller会发起选举。
环境配置
配置虚拟机
首先创建一个虚拟机，并克隆三台虚拟机
登录101虚拟机，进行配置端口
vim /etc/sysconfig/network-scripts/ifcfg-ens33
修改IPADDR
IPADDR="192.168.27.101"
配置主机名
vim /etc/hostname
修改主机名为centos101
centos101
然后reboot
reboot
其余两台虚拟机按照101的配置进行设置
安装JDK
在101虚拟机安装JDK
首先查看虚拟机有没有自带的jdk，如果有需要先删除再安装jdk
rpm -qa | grep -i java
获取jdk安装包
wget --no-check-certificate --no-cookies --header "Cookie: oraclelicense=accept-securebackup-cookie" http://download.oracle.com/otn-pub/java/jdk/8u131-b11/d54c1d3a095b4ff2b6607d096fa80163/jdk-8u131-linux-x64.tar.gz
解压安装包
tar -zxvf jdk-8u131-linux-x64.tar.gz
配置环境变量
sudo vim /etc/profile.d/my_env.sh
在my_env.sh中输入以下内容
#JAVA_HOME
export JAVA_HOME=/mydata/jdk/jdk1.8.0_131
export PATH=$PATH:$JAVA_HOME/bin
执行source命令
source /etc/profile
查看jdk是否安装完毕
java -version
安装Hadoop
使用华为云开源镜像获取安装包
https://mirrors.huaweicloud.com/home
wget https://mirrors.huaweicloud.com/apache/hadoop/core/hadoop-3.1.3/hadoop-3.1.3.tar.gz
解压安装包
tar -zxvf hadoop-3.1.3.tar.gz
配置环境变量
sudo vim /etc/profile.d/my_env.sh
加入Hadoop的环境变量
#HADOOP_HOME
export HADOOP_HOME=/mydata/hadoop/hadoop-3.1.3
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin
执行source
source /etc/profile
查看是否安装完成
hadoop version
配置其余两台虚拟机
scp安全拷贝命令将安装好的jdk和hadoop拷贝到102和103
scp -r jdk/ root@192.168.27.102:/mydata
scp -r hadoop/ root@192.168.27.102:/mydata
注意：rsync远程同步工具可以只对差异文件做更新，而scp是复制所有文件
xsync集群分发脚本可以循环复制文件所有的节点的相同目录下
为了使用xsync，需要建立一个脚本来实现，脚本需要放在声明了全局变量的路径下，因此修改之前创建的my_env.sh，将/mydata/bin声明全局变量
#MYDATA_HOME
export MYDATA_HOME=/mydata
export PATH=$PATH:$MYDATA_HOME/bin
[root@centos101 mydata]# source /etc/profile
[root@centos101 mydata]# echo $PATH
/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin:/mydata/jdk/jdk1.8.0_131/bin:/mydata/jdk/jdk1.8.0_131/bin:/mydata/hadoop/hadoop-3.1.3/bin:/mydata/hadoop/hadoop-3.1.3/sbin:/mydata/jdk/jdk1.8.0_131/bin:/mydata/hadoop/hadoop-3.1.3/bin:/mydata/hadoop/hadoop-3.1.3/sbin:/mydata/bin
可以看到现在已经加入成功，进入/mydata/bin,新建一个xsync脚本
[root@centos101 mydata]# cd bin/
[root@centos101 bin]# ls
[root@centos101 bin]# vim xsync
#!/bin/bash#1. 判断参数个数
if [ $# -lt 1 ]
thenecho Not Enough Arguement!exit;
fi#2. 遍历集群所有机器
for host in 192.168.27.101 192.168.27.102 192.168.27.103
doecho ==================== $host ====================#3. 遍历所有目录，挨个发送for file in $@do#4. 判断文件是否存在if [ -e $file ]then#5. 获取父目录pdir=$(cd -P $(dirname $file); pwd)#6. 获取当前文件的名称fname=$(basename $file)ssh $host "mkdir -p $pdir"rsync -av $pdir/$fname $host:$pdirelseecho $file does not exists!fidone
done
配置脚本可执行权限
chmod 777 xsync
然后就可以直接使用xsync将新增的文件同步到其他主机，比如刚刚在101上面建立的/mydata/bin
xsync bin/
如果遇到rsync未找到命令可以先安装，同步和被同步的主机都需要安装这个命令
yum -y install rsync
完成之后将配置环境变量的文件my_env.sh同步到其他主机
进入到my_env.sh文件所在目录下执行命令
xsync my_env.sh
同步完成后，将刚同步的主机执行source命令即可
配置SSH免密
首先先ssh 访问别的主机，生成.ssh隐藏文件
ssh 192.168.27.102
然后可以在/root目录下，通过ls -al查看隐藏文件
ls -al
进入.ssh,生成公钥和私钥，输入命令执行三次回车即可
ssh-keygen -t rsa
可以查看公钥
cat id_rsa.pub
将公钥拷贝给其他主机
ssh-copy-id 192.168.27.102
这样就可以不需要密码访问别的主机了（自己也可以配置一下免密访问自己）
可以用xsync命令来测试一下，如果不需要使用密码就可以同步文件到其他主机则设置成功
安装Kafka
首先在官网下载Kafka
https://kafka.apache.org/downloads
然后将安装包放入/mydata/kafka
如果需要直接拖入文件到虚拟机，可以提前执行命令下载相应组件
yum -y install lrzsz
[root@localhost kafka]# yum -y install lrzsz
解压压缩包
tar -zxvf kafka_2.12-3.0.0.tgz
进入解压后的文件夹，配置kafka信息
[root@localhost kafka]# cd kafka_2.12-3.0.0
[root@localhost kafka_2.12-3.0.0]# ls
bin config libs LICENSE licenses NOTICE site-docs
[root@localhost kafka_2.12-3.0.0]# cd config
注意：如果发现vim:未找到命令情况
[root@localhost config]# vim server.properties -bash: vim: 未找到命令显示未找到命令，则查看有什么安装包没有安装
[root@localhost config]# rpm -qa | grep vim vim-minimal-7.4.629-7.el7.x86_64正常情况下应该有以下安装包
[root@localhost config]# rpm -qa | grep vim vim-minimal-7.4.629-8.el7_9.x86_64 vim-common-7.4.629-8.el7_9.x86_64 vim-filesystem-7.4.629-8.el7_9.x86_64 vim-enhanced-7.4.629-8.el7_9.x86_64将缺少的安装包安装上即可
yum -y install vim-enhanced
执行vim命令，查看server.properties
vim server.properties
其中有几个地方需要注意
-
broker.id=0 相当于kafka在集群中的唯一标识，不同kafka必须要不同，不能重复
-
log.dirs=/tmp/kafka-logs 日志的路径默认是在临时文件夹下，要进行修改，将文件放到kafka的安装目录中
log.dirs=/mydata/kafka/datas -
zookeeper.connect=localhost:2181 配置zookeeper的链接地址,后面加/kafka是为了规范管理kafka在zk中存储的路径
zookeeper.connect=192.168.27.101:2181,192.168.27.102:2181,192.168.27.103:2181/kafka
将kafka文件夹分发到其他主机
xsync kafka/
然后修改其他主机的brokerid,依次设置为0，1，2
最后设置环境变量
vim /etc/profile.d/my_env.sh
#KAFKA_HOME
export KAFKA_HOME=/mydata/kafka/kafka_2.12-3.0.0
export PATH=$PATH:$KAFKA_HOME/bin
修改完之后执行source命令
source /etc/profile
分发到其他主机上，其他主机也需要执行source命令
xsync /etc/profile.d/my_env.sh
启动Kafka
先启动zk，然后在启动Kafka，zk启动使用之前在Zookeeper中编写的zk.sh脚本
zk.sh start
启动kafka，三台主机都要执行
bin/kafka-server-start.sh -daemon config/server.properties
查看是否启动成功
[root@centos101 kafka_2.12-3.0.0]# jps
1954 QuorumPeerMain
3605 Kafka
3641 Jps
为了方便集群控制，编写脚本
#!/bin/bash
case $1 in
"start")for i in 192.168.27.101 192.168.27.102 192.168.27.103doecho "-------------启动 $i kafka-------------"ssh $i "/mydata/kafka/kafka_2.12-3.0.0/bin/kafka-server-start.sh -daemon /mydata/kafka/kafka_2.12-3.0.0/config/server.properties"done
;;
"stop")for i in 192.168.27.101 192.168.27.102 192.168.27.103doecho "-------------停止 $i kafka-------------"ssh $i "/mydata/kafka/kafka_2.12-3.0.0/bin/kafka-server-stop.sh"done
;;
esac
在/mydata/bin（自己配的具备环境变量的目录，专门存放脚本文件）下新建kf.sh，写入脚本
vim kf.sh
配置权限
chmod 777 kf.sh
[root@centos101 bin]# ll
总用量 12
-rwxrwxrwx. 1 root root 480 7月 18 15:50 kf.sh
-rwxrwxrwx. 1 root root 766 7月 14 15:10 xsync
-rwxrwxrwx. 1 root root 674 7月 17 14:40 zk.sh
注意：如果要关闭时，要先关闭Kafka再关闭Zookeeper，并且要确认kafka已经完全关闭。
文件丢失或损坏问题的终极指南：从基础到高级修复技巧)
