🙆♂️🙆♂️ 写在前面
🏠 个人主页: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,所以这里我们一般不用做修改
版权归原作者 it春和 所有, 如有侵权,请联系我们删除。