当前位置： 首页 > news >正文

Kafka入门-集群基础环境搭建（JDK/Hadoop 部署 + 虚拟机配置 + SSH 免密+Kafka安装启动）

news 来源：原创 2025/6/6 18:16:28

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已经完全关闭。

相关文章：

  • 乾元通渠道商中标西藏2024年应急装备采购配置项目
  • SSM 框架核心知识详解（Spring + SpringMVC + MyBatis）
  • Sql Server 中常用语句
  • 数据结构第八章（一） 插入排序
  • 李飞飞World Labs开源革命性Web端3D渲染器Forge！3D高斯溅射技术首次实现全平台流畅运行
  • GPUCUDA 发展编年史：从 3D 渲染到 AI 大模型时代（上）
  • Elasticsearch的审计日志（Audit Logging）介绍
  • 【Android】RV折叠适配器
  • 王道入门50题答案
  • OpenLayers 地图标注之聚合标注
  • 微信小程序前端面经
  • 搭建强化推荐的决策服务架构
  • Quick UI 组件加载到 Axure
  • HashMap中的put方法执行流程（流程图）
  • 搭建nginx的负载均衡
  • Rust学习（1）
  • UR 协作机器人「三剑客」：精密轻量担当（UR7e）、全能协作主力（UR12e）、重型任务专家（UR15）
  • flask功能使用总结和完整示例
  • 解决Java项目NoProviderFoundException报错
  • 日志项目——日志系统框架设计
  • 沈阳网站建设的公司哪家好/国内高清视频素材网站推荐
  • 网站建设公司前十名/如何进行网络推广营销
  • 云南SEO网站建设/软文网官网
  • 网站建设网站定制/seo建站
  • wordpress版权图片/邯郸seo营销
  • 做服装搭配直接售卖的网站/网络新闻发布平台