kafka 基本操作命令

Posted 2021-03-04 ryjjava

• 查看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列表
root@root:/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是消费滞后区间
root@root:/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包中执行指定的工具类.

 

root@root:/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.

root@root:/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
root@root:/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
root@root:/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