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

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