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

kafka部署以及常用命令详细总结

news 2025/7/12 15:56:24

1环境准备

1.1ip规划

ip: 192.168.1.200

1.2配置主机名

#设置主机名
hostnamectl set-hostname node1

1.3配置hosts

[root@node1 ~]# cat >> /etc/hosts << 'EOF'192.168.1.200 node1
EOF

2部署

2.1安装包准备

将以下安装包从官网下载到本地

jdk-8u371-linux-x64.tar.gz

apache-zookeeper-3.7.1-bin.tar.gz

kafka_2.12-3.4.0.tgz

上述安装包已整理上传到我的资源，需要可自行下载

#将下载好的安装包上传到服务器
[root@node1 ~]# ll
总用量 252268
-rw-------. 1 root root      1257 3月  29 2023 anaconda-ks.cfg
-rw-r--r--  1 root root  12649765 7月  12 15:04 apache-zookeeper-3.7.1-bin.tar.gz
-rw-r--r--  1 root root 139219380 7月  12 15:10 jdk-8u371-linux-x64.tar.gz
-rw-r--r--  1 root root 106441367 7月  12 15:04 kafka_2.12-3.4.0.tgz

2.2安装jdk

#解压安装包
[root@node1 ~]# tar -xvf jdk-8u371-linux-x64.tar.gz -C /usr/local/#配置环境变量
[root@node1 ~]# cat >> /etc/profile << 'EOF'#Set for jdk
export JAVA_HOME=/usr/local/jdk1.8.0_371
export PATH=$PATH:$JAVA_HOME/bin
EOF[root@node1 ~]# source /etc/profile
[root@node1 ~]# java -version
openjdk version "1.8.0_262"
OpenJDK Runtime Environment (build 1.8.0_262-b10)
OpenJDK 64-Bit Server VM (build 25.262-b10, mixed mode)

2.3部署zookeeper

[root@node1 ~]# tar -xf apache-zookeeper-3.7.1-bin.tar.gz -C /usr/local/
[root@node1 ~]# cd /usr/local/
[root@node1 local]# mv apache-zookeeper-3.7.1-bin/ zookeeper
[root@node1 local]# cd zookeeper/conf/[root@node1 conf]# cp zoo_sample.cfg zoo.cfg[root@node1 conf]# vi zoo.cfg
dataDir=/usr/local/zookeeper/data
#server.1=node1:2888:3888
#server.2=node1:2888:3888
...
#多节点需要server.1= 配置，用于多节点之间的心跳检测和选举机制#配置myid，myid 文件用于标识 ZooKeeper 服务器的身份，即使是在单节点环境中，ZooKeeper 也需要知道自己是谁。内容应该是数字 1，表示这是一个独立的单节点部署。[root@node1 conf]# mkdir /usr/local/zookeeper/data
[root@node1 conf]# echo 1 > /usr/local/zookeeper/data/myid#设置环境变量
[root@node1 conf]# cd
[root@node1 ~]# cat >> .bash_profile << 'EOF'#set for zookeeper
export ZOOKEEPER_HOME=/usr/local/zookeeper
export PATH=$PATH:$ZOOKEEPER_HOME/bin
EOF[root@node1 ~]# source .bash_profile#启动zookeeper
[root@node1 ~]# zkServer.sh start[root@node1 ~]# jps
2038 Jps
1995 QuorumPeerMain

2.4部署kafka

[root@node1 ~]# tar -xf kafka_2.12-3.4.0.tgz -C /usr/local/
[root@node1 ~]# cd /usr/local/
[root@node1 local]# mv kafka_2.12-3.4.0/ kafka
[root@node1 local]# cd kafka/config/[root@node1 config]# vi server.properties
listeners=PLAINTEXT://node1:9092
log.dirs=/usr/local/kafka/logs
zookeeper.connect=node1:2181[root@node1 config]# mkdir /usr/local/kafka/logs#设置环境变量
[root@node1 config]# cd
[root@node1 ~]# cat >> .bash_profile << 'EOF'#set for kafka
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
EOF[root@node1 ~]# source .bash_profile#启动
[root@node1 ~]# kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties[root@node1 ~]# jps
1995 QuorumPeerMain
2412 Kafka
2429 Jps

3kafka常用使用命令

3.1创建topic以及给用户授权

#创建topic
[root@node1 ~]# kafka-topics.sh --bootstrap-server 192.168.1.200:9092 --create --replication-factor 1 --partitions 1 --topic my-topic
Created topic my-topic.#给用户 mrloam 授予生产权限
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --add --allow-principal User:mrloam --producer --topic my-topic
Warning: support for ACL configuration directly through the authorizer is deprecated and will be removed in a future release. Please use --bootstrap-server instead to set ACLs through the admin client.
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=my-topic, patternType=LITERAL)`: (principal=User:mrloam, host=*, operation=CREATE, permissionType=ALLOW)(principal=User:mrloam, host=*, operation=WRITE, permissionType=ALLOW)(principal=User:mrloam, host=*, operation=DESCRIBE, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=my-topic, patternType=LITERAL)`: (principal=User:mrloam, host=*, operation=CREATE, permissionType=ALLOW)(principal=User:mrloam, host=*, operation=WRITE, permissionType=ALLOW)(principal=User:mrloam, host=*, operation=DESCRIBE, permissionType=ALLOW) #创建消费组并授权，消费组不存在则授权消费权限报错
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.200:9092 --topic my-topic --group my-group#给用户 mrloam 授予消费权限，指定消费组 my-group
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --add --allow-principal User:mrloam --consumer --topic my-topic --group my-group#给用户 mrloam 授予消费权限，所有消费组
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --add --allow-principal User:mrloam --consumer --topic my-topic --group *
#查看所有topic
[root@node1 ~]# kafka-topics.sh --bootstrap-server 192.168.1.200:9092 --list#查看指定topic的详细信息
[root@node1 ~]# kafka-topics.sh --bootstrap-server 192.168.1.200:9092 --topic my-topic --describe

3.2生产、消费

#生产
[root@node1 ~]# kafka-console-producer.sh --bootstrap-server 192.168.1.200:9092 --topic my-topic
#创建或指定 groupid 进行消费
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.200:9092 --topic my-topic --group my-group
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.200:9092 --topic my-topic --consumer-property --group.id=my-group#将偏移量移动到最早时间位置
[root@node1 ~]# kafka-consumer-groups.sh --bootstrap-server 192.168.1.200:9092 --group my-group --reset-offsets --topic my-topic --to-earliest --execute#设置偏移量到最新时间
[root@node1 ~]# kafka-consumer-groups.sh --bootstrap-server 192.168.1.200:9092 --group my-group --topic my-topic --reset-offsets --to-latest --execute#移动偏移量到指定时间位置
[root@node1 ~]# kafka-consumer-groups.sh --bootstrap-server 192.168.1.200:9092 --group my-group --topic my-topic --reset-offsets --to-datetime "2022-12-08T02:00:00.000" --execute --command-config consumer.properties#移动偏移量到指定位置 200 处
[root@node1 ~]# kafka-consumer-groups.sh --bootstrap-server 192.168.1.200:9092 --group my-group --topic my-topic --reset-offsets --to-offset 200 --execute

#显示key消费,消费出的消息将打印出消息体的key和value
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.200:9092 --topic my-topic --group my-group --property print.key=true#指定消费10条数据
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.200:9092 \
--topic my-topic \
--group my-group \
--max-messages 10 \
--property print.offset=true \
--property print.partition=true \
--property print.headers=true \
--property print.timestamp=true \
--consumer.config consumer.properties#过滤指定时间的数据, 时间戳: $(date -d '2024-07-11 15:20:00.000' +%s%3N)
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server 192.168.1.200:9092 \
--topic my-topic \
--group my-group \
--property print.timestamp=true \
--consumer.config consumer.properties \
--from-beginning | awk -F 'CreateTime:' '$2>=1675742400000 && $2<=1675749600000 {print $0}'

3.3查看，添加，移除权限

#查看topic权限
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --list --topic my-topic#添加权限
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --add --allow-principal User:mrloam --allow-host * --producer --topic my-topic
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --add --allow-principal User:mrloam --allow-host 192.168.1.* --producer --topic my-topic
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --add --allow-principal User:mrloam --allow-host * --consumer --topic my-topic --group my-group
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --add --allow-principal User:mrloam --allow-host * --consumer --topic my-topic --group *

#移除权限
[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --remove --allow-principal User:mrloam --allow-host 192.168.1.* --producer --topic my-topic --force[root@node1 ~]# kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.200:2181 --remove --allow-principal User:mrloam --allow-host 192.168.1.* --consumer --group my-group --topic my-topic --force

增加topic分区数

[root@node1 ~]# kafka-topics.sh --alter --bootstrap-server 192.168.1.200:9092 --topic my-topic --partitions 3

消费组信息查询

#查看所有消费组
[root@node1 ~]# kafka-consumer-groups.sh --list --bootstrap-server 192.168.1.200:9092#查看消费组详细信息
[root@node1 ~]# kafka-consumer-groups.sh --describe --bootstrap-server 192.168.1.200:9092 --group my-group

删除topic

[root@node1 ~]# kafka-topics.sh --delete --bootstrap-server 192.168.1.200:9092 --topic my-topic

#查看kafka版本
[root@node1 ~]# kafka-topics.sh --version
[root@node1 ~]# kafka-server-start.sh --version[root@node1 ~]# cd $KAFKA_HOME
[root@node1 ~]# find ./libs -name 'kafka_*'|grep -o 'kafka[^\\n]*'	#kafka_2.12-3.4.0.jar #scala版本:2.11   kafka版本:3.4.0#查看zk版本
[root@node1 ~]# zkServer.sh version#连接到 ZooKeeper 服务器
[root@node1 ~]# zkCli.sh
version[root@node1 ~]# echo stat | nc 192.168.1.200 2181
http://www.lryc.cn/news/399312.html

相关文章：

  • 代码随想录算法训练营第29天|LeetCode 134. 加油站、135. 分发糖果、860.柠檬水找零、406.根据身高重建队列
  • 代理模式(大话设计模式)C/C++版本
  • 本人学习保存-macOS打开Navicat提示「“Navicat Premium”已损坏，无法打开。 你应该将它移到废纸篓。」的解决方法
  • 《Cross-Image Pixel Contrasting for Semantic Segmentation》论文解读
  • 技术周总结 2024.07.08~07.14(算法，Python,Java,Scala,PHP)
  • UnityECS学习中问题及总结entityQuery.ToComponentDataArray和entityQuery.ToEntityArray区别
  • [python]基于yolov10+gradio目标检测演示系统设计
  • 浏览器开发者视角及CSS表达式选择元素
  • GuLi商城-商品服务-API-品牌管理-统一异常处理
  • VUE+Spring Flux实现SSE长连接
  • C#实现Winform程序右下角弹窗消息提示
  • Java三剑客：封装、继承、多态的魔法世界
  • 0145__Linux的capability
  • # Redis 入门到精通（一）数据类型（4）
  • 西邮计科嵌入式复习
  • Java如何使用 HttpClientUtils 发起 HTTP 请求
  • 无人机的工作原理
  • 敏捷开发笔记(第10章节)--Liskov原则（LSP）
  • 基于SSM的校园一卡通管理系统的设计与实现
  • 新版Android Studio中设置gradle的JDK版本
  • 打造你的智能家居指挥中心：基于STM32的多协议（zigbee、http）网关（附代码示例）
  • 【基于R语言群体遗传学】-16-中性检验Tajima‘s D及连锁不平衡 linkage disequilibrium (LD)
  • 防火墙组网与安全策略实验
  • xmind梳理测试点，根据这些测试点去写测试用例
  • MICCAI 2024 每日一篇论文 纯纯直读 CUTS:用于多粒度无监督医学图像分割的深度学习和拓扑框架
  • 实验9 存储过程与函数的创建管理实验
  • 计算机网络--tcpdump和iptable设置、内核参数优化策略
  • Vue3框架搭建2：axios+typescript封装
  • 【机器学习】使用决策树分类器预测汽车安全性的研究与分析
  • 【香橙派 Orange pi AIpro】| 开发板深入使用体验