kafka基本操作

Posted 2021-01-18 654wangzai321

1. 启动kafka

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

2. 创建topic

./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

3. 查询topic

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

4. 查询topic的详细描述

./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

5. 发送数据

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

6. 接收数据

./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
## --from-beginning表示从头开始读，如果不加的话，从启动consumer开始接收数据

7. 删除topic

./bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test
 注意：如果kafaka启动时加载的配置文件中server.properties没有配置delete.topic.enable=true，那么此
时的删除并不是真正的删除，而是把topic标记为：marked for deletion
– 此时你若想真正删除它，可以登录zookeeper客户端，进入终端后，删除相应节点
zkCli.sh 登录进去之后
ls /brokers/topics  查看topic
rmr /brokers/topics/test 把对应的topic删除

8. 查询topic的offset的范围

用下面命令可以查询到topic:test broker:master:9092的offset的最小值：
./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list master:9092 --topic test --time -2
查询offset的最大值：
./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list master:9092 --topic test --time -1