0


一文学会基于发布订阅模式的消息队列Kafka的安装部署和基本使用

🙆‍♂️🙆‍♂️ 写在前面

​🏠​ 个人主页:csdn春和
📚 推荐专栏:更多专栏尽在主页!
 Scala专栏(spark必学语言 已完结)
 JavaWeb专栏(从入门到实战超详细!!!)
 SSM专栏 (更新中…)
​📖​ 本期文章:一文学会基于发布订阅模式的消息队列Kafka的安装部署和基本使用


如果对您有帮助还请三连支持,定会一 一回访!🙋🏻‍♂️


📌本文目录


一、Kafka安装部署

前提:需要安装zookeeper集群
kafka最新版本已经可以摒弃zookeeper,实现自身管理,但是这里我安装的kafka版本是2.7.0,所以还是需要提前安装好zookeeper集群环境。
zookeeper集群安装参考文章:zookeeper的集群安装部署

1.1、kafka集群规划

hadoop02hadoop03hadoop04zkzkzkkafkakafkakafka

1.2、安装详细步骤

1、下载kafka的安装包

kafka下载地址 点我,请轻一点

2、上传到linux并解压到指定的目录

在这里插入图片描述

tar -zxvf kafka_2.12-2.7.0.tgz -C /opt/module/

3、修改名称

mv kafka_2.12-2.7.0/ kafka-2.7.0

在这里插入图片描述

4、在/opt/module/kafka-2.7.0 目录下创建 logs 文件夹

在这里插入图片描述

5、修改配置文件 server.properties

#broker 的全局唯一编号,不能重复 
broker.id=0#删除 topic 功能打开使能
delete.topic.enable=true
#处理网络请求的线程数量 
num.network.threads=3#用来处理磁盘 IO 的现成数量 
num.io.threads=8#发送套接字的缓冲区大小 
socket.send.buffer.bytes=102400#接收套接字的缓冲区大小 
socket.receive.buffer.bytes=102400#请求套接字的缓冲区大小 
socket.request.max.bytes=104857600#kafka 运行日志存放的路径 
log.dirs=/opt/module/kafka-2.7.0/logs 
#topic 在当前 broker 上的分区个数 默认分区数 
num.partitions=1#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1#segment 文件保留的最长时间,超时将被删除 
log.retention.hours=168#配置连接 Zookeeper 集群地址 
zookeeper.connect=hadoop02:2181,hadoop03:2181,hadoop04:2181

6、配置环境变量

sudo vim /etc/profile.d/my_env.sh
#KAFKA_HOMEexportKAFKA_HOME=/opt/module/kafka-2.7.0
exportPATH=$PATH:$KAFKA_HOME/bin

在这里插入图片描述

在这里插入图片描述

7、同步分发到其他机器

分发之后要记得配置其他机器上的环境变量

xsync kafka-2.7.0/ 

在这里插入图片描述

8、分别在hadoop03 hadoop04上修改配置文件server.properties

修改broker.id 使得id不重复

这里如果id重复了 那么启动的kafka进程会自动挂掉

在这里插入图片描述

9、启动集群

注意这里我们需要先启动zookeeper集群

然后依次在 hadoop02、hadoop03、hadoop04 节点上启动 kafka

bin/kafka-server-start.sh -daemon config/server.properties

在这里插入图片描述

10、关闭集群

关闭集群需要在每一台机器上关闭kafka

[zhou@hadoop02 kafka-2.7.0]$ bin/kafka-server-stop.sh stop 
[zhou@hadoop03 kafka-2.7.0]$ bin/kafka-server-stop.sh stop 
[zhou@hadoop04 kafka-2.7.0]$ bin/kafka-server-stop.sh stop

11、kafka集群脚本

上述我们开启kafka集群或者关闭都很麻烦,需要在三台机器上分别执行命令

创建一个自定义脚本,我的一般都放在/home/zhou/bin目录下

#! /bin/bashcase$1in"start"){foriin hadoop02 hadoop03 hadoop04
 doecho" --------启动 $i 的Kafka-------"ssh$i"/opt/module/kafka-2.7.0/bin/kafka-server-start.sh -daemon /opt/module/kafka-2.7.0/config/server.properties"done};;"stop"){foriin hadoop02 hadoop03 hadoop04
 doecho" --------停止 $i 的Kafka-------"ssh$i"/opt/module/kafka-2.7.0/bin/kafka-server-stop.sh "done};;esac

在这里插入图片描述

在这里插入图片描述

测试脚本:

启动集群:kf.sh start

停止集群:kf.sh stop

注意:停止 Kafka 集群时,一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper集群。因为 Zookeeper 集群当中记录着 Kafka 集群相关信息,Zookeeper 集群一旦先停止, Kafka 集群就没有办法再获取停止进程的信息,只能手动杀死 Kafka 进程了

在这里插入图片描述

二、kafka命令行操作

2.1、基本主题命令

1、创建topic

bin/kafka-topics.sh --zookeeper hadoop02:2181 --create --replication-factor 3 --partitions 1 --topic first 

选项说明: 
--topic 定义topic的名字 
--replication-factor 定义副本数 
--partitions 定义分区数

在这里插入图片描述

2、查看当前服务器中所有的topic

方式一:

bin/kafka-topics.sh --zookeeper hadoop02:2181 --list

方式二:

bin/kafka-topics.sh --bootstrap-server hadoop02:9092 --list 

在这里插入图片描述

3、查看某个topic的详情

bin/kafka-topics.sh --zookeeper hadoop02:2181 --describe --topic first

bin/kafka-topics.sh --bootstrap-server hadoop02:9092 --describe --topic first

在这里插入图片描述

4、修改分区数(

注意:分区数只能增加,不能减少

bin/kafka-topics.sh --bootstrap-server hadoop02:2181 --alter --partitions 2 --topic first

bin/kafka-topics.sh --bootstrap-server hadoop02:9092 --alter --topic first --partitions 3

在这里插入图片描述

5、创建主题的时候 副本数不能超过brokers的数量

在这里插入图片描述

6、删除topic

bin/kafka-topics.sh --zookeeper hadoop02:2181 --delete --topic first

bin/kafka-topics.sh --bootstrap-server hadoop02:9092 --delete --topic first

在这里插入图片描述

2.2、命令行控制台生产者消费者测试

0、环境准备

创建一个主题

in/kafka-topics.sh --bootstrap-server hadoop02:9092 --create --replication-factor 3 --partitions 1 --topic test

1、启动生产者

bin/kafka-console-producer.sh --broker-list hadoop02:9092 --topic test

bin/kafka-console-producer.sh --bootstrap-server hadoop02:9092 --topic test

在这里插入图片描述

2、启动消费者

#bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first 过时了 现在已经没有这个命令了 # 使用下面两个命令 
bin/kafka-console-consumer.sh --bootstrap-server hadoop02:9092 --topic first bin/kafka-console-consumer.sh --bootstrap-server hadoop02:9092 --topic first --from-beginning

在这里插入图片描述

2.3、数据日志分离

将kafka的数据和日志分开存储

【1】停止kafka集群 zk

【2】删除所有机器上的logs目录

【3】删除zookeeper上所有与kafka相关的节点信息
这里为了方便我是用了zookeeper的图形化管理工具,preetyZoo
在这里插入图片描述

【4】创建data目录

在这里插入图片描述

【5】修改config下的server.properties文件内容

在这里插入图片描述

【6】重新启动kafka集群
在这里插入图片描述

在这里插入图片描述

2.4、kafka的默认副本和默认分区

思考一个问题:当我们生产者往一个不存在的主题上发数据的时候能不能发送成功?会不会报错?

在这里插入图片描述

默认一个副本一个分区 可以修改配置文件 server.properties

在这里插入图片描述

但是一般我们都会先创建好topic,所以这里我们一般不用做修改


本文转载自: https://blog.csdn.net/m0_66947873/article/details/124921573
版权归原作者 it春和 所有, 如有侵权,请联系我们删除。

“一文学会基于发布订阅模式的消息队列Kafka的安装部署和基本使用”的评论:

还没有评论