- 查看kafka版本
进入kafka安装目录 ... kafka/libs,看到类似kafka_2.12-2.0.0.jar这样的文件,2.12为scala版本,2.0.0是kafka版本(kafka使用了Scala进行开发).
- zookeeper-server-start.sh
参照 kafka环境搭建
- zookeeper-server-stop.sh
停止kafka bin/kafka-server-stop.sh
- kafka-server-start.sh
参照 kafka环境搭建
- kafka-server-stop.sh
停止zookeeper bin/kafka-server-stop.sh
- kafka-topics.sh
参照 kafka环境搭建
创建topic bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test 展示topic bin/kafka-topics.sh --list --zookeeper localhost:2181 描述topic bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic 删除topic bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topicname 删除topic中存储的内容 在config/server.properties中找到如下的位置log.dirs=路径,删除log.dirs指定的文件目录,然后重新启动就可以了
- kafka-consumer-groups.sh
-- 获取group列表[email protected]:/usr/local/kafka/kafka_2.12-2.0.0$ bin/kafka-consumer-groups.sh --bootstrap-server 192.168.101.75:6667 --list console-consumer-61307 -- 查看具体的消费者group信息。其中CURRENT-OFFSET是该分区当前消费到的offset;LOG-END-OFFSET是该分区当前最新的offset;LAG是消费滞后区间[email protected]:/usr/local/kafka/kafka_2.12-2.0.0$ bin/kafka-consumer-groups.sh --bootstrap-server 192.168.101.75:6667 --group console-consumer-61307 --describe Consumer group ‘console-consumer-61307‘ has no active members. TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID test 0 3 3 0 - - -
--重置消费者offset:指定GROUP_NAME和topic的offset修改到NEW_OFFSET的位置,重启消费者后,消费中将从指定的offset处消费。注意这里只能NEW_OFFSET只能设置一个值,也就是说,所有的分区都将使用这个值,如果分区消息负载不均衡,需要考虑是否适用
bin/kafka-consumer-groups.sh --bootstrap-server BORKER_HOST1:PORT1,BORKER_HSOT2:PORT2 --group GROUP_NAME --reset-offsets --execute --to-offset NEW_OFFSET --topic TOPIC_NAME
--重置消费者offset:将指定GROUP_NAME和topic的offset修改到earliest或者latest位置,使得消费者从头或者从尾部消费
bin/kafka-consumer-groups.sh --bootstrap-server BORKER_HOST1:PORT1,BORKER_HSOT2:PORT2 --group GROUP_NAME --reset-offsets --execute --to-earliest/--to-latest --topic TOPIC_NAME
- kafka-console-consumer.sh
参照 kafka环境搭建
-- 从头消费 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo --from-beginning -- 指定offset消费,并限制消费数量(partition必须指定,除非‘--offset‘是明确的) bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic demo --offset latest --partition 0 --max-messages 2
- kafka-run-class.sh
从kafka jar包中执行指定的工具类.
[email protected]:/usr/local/kafka/kafka_2.12-2.0.0# bin/kafka-run-class.sh kafka.tools.GetOffsetShell An interactive shell for getting topic offsets. Option Description ------ ----------- --broker-list <String: hostname: REQUIRED: The list of hostname and port,...,hostname:port> port of the server to connect to. --max-wait-ms <Integer: ms> DEPRECATED AND IGNORED: The max amount of time each fetch request waits. (default: 1000) --offsets <Integer: count> DEPRECATED AND IGNORED: number of offsets returned (default: 1) --partitions <String: partition ids> comma separated list of partition ids. If not specified, it will find offsets for all partitions (default: ) --time <Long: timestamp/-1(latest)/-2 timestamp of the offsets before that (earliest)> (default: -1) --topic <String: topic> REQUIRED: The topic to get offset from. [email protected]:/usr/local/kafka/kafka_2.12-2.0.0# bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.101.75:6667 --topic test test:0:3 [email protected]:/usr/local/kafka/kafka_2.12-2.0.0# bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.101.75:6667 --topic test --time -1 test:0:3 [email protected]:/usr/local/kafka/kafka_2.12-2.0.0# bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.101.75:6667 --topic test --time -2 test:0:0
原文地址:https://www.cnblogs.com/ryjJava/p/12545617.html