`
kavy
  • 浏览: 868293 次
  • 性别: Icon_minigender_1
  • 来自: 上海
社区版块
存档分类
最新评论

kafka监控命令kafka-run-class.sh查看消费了多少条数据

 
阅读更多

kafka自带了很多工具类,在源码kafka.tools里可以看到:


源码包下载地址:http://archive.apache.org/dist/kafka/

 

这些类该如何使用呢,kafka的设计者早就为我们考虑到了,在${KAFKA_HOME}/bin下,有很多的脚本,其中有一个kafka-run-class.sh,通过这个脚本,可以调用其中的tools的部分功能,如调用kafka.tools里的ConsumerOffsetChecker.scala


 

查看consumer组内消费的offset
./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper **:2181 --group ** --topic **

 

其中的group可去zookeeper中查看:
[hadoop@h71 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh
[zk: localhost:2181(CONNECTED) 2] ls /consumers
[console-consumer-74653, WordcountConsumerGroup]
(console-consumer-74653这个组当我在另一个窗口启动消费者[hadoop@h71 kafka_2.10-0.8.2.0]$ bin/kafka-console-consumer.sh --zookeeper h71:2181,h72:2181,h73:2181 --topic test --from-beginning时才会有,关闭消费者进程该组会自动消失)

 

我们在使用kafka消费信息的过程中,不同group的consumer是可以消费相同的信息的,group是在创建consumer时指定的,如果group不存在,会自动创建。其实简单点说就是每个group都会在zk中注册,区别就是注册过还是没注册过。每个group内的consumer只能消费在group注册过之后生产的信息。

 

执行结果如下:列出了所有消费者组的所有信息,包括Group(消费者组)、Topic、Pid(分区id)、Offset(当前已消费的条数)、LogSize(总条数)、Lag(未消费的条数)、Owner

[plain] view plain copy
 
  1. [hadoop@h71 kafka_2.10-0.8.2.0]$ bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper 192.168.8.71:2181,192.168.8.72:2181,192.168.8.73:2181 --group WordcountConsumerGroup --topic test  
  2. 或者:  
  3. [hadoop@h71 kafka_2.10-0.8.2.0]$ bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper=192.168.8.71:2181,192.168.8.72:2181,192.168.8.73:2181 --group=WordcountConsumerGroup --topic=test  
  4. 再或者:  
  5. [hadoop@h71 kafka_2.10-0.8.2.0]$ bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper=h71:2181 --group=WordcountConsumerGroup  
  6. Group           Topic                          Pid Offset          logSize         Lag             Owner  
  7. test-consumer-group test                           0   4               40              36              none  
  8. test-consumer-group test                           1   14              57              43              none  
  9.   
  10. [hadoop@h71 kafka_2.10-0.8.2.0]$ bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper 192.168.8.71:2181,192.168.8.72:2181,192.168.8.73:2181 --group=WordcountConsumerGroup  
  11. Group           Topic                          Pid Offset          logSize         Lag             Owner  
  12. WordcountConsumerGroup test                           0   9               40              31              none  
  13. WordcountConsumerGroup test                           1   21              57              36              none  
  14. [hadoop@h71 kafka_2.10-0.8.2.0]$ bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper 192.168.8.71:2181,192.168.8.72:2181,192.168.8.73:2181 --group=console-consumer-42639  
  15. Group           Topic                          Pid Offset          logSize         Lag             Owner  
  16. console-consumer-42639 test                           0   40              40              0               console-consumer-42639_h71-1498000434797-f1c703cf-0  
  17. console-consumer-42639 test                           1   57              57              0               console-consumer-42639_h71-1498000434797-f1c703cf-0  

 

 

细看kafka-run-class.sh脚本,它是调用了ConsumerOffsetChecker的main方法,所以,我们也可以通过java代码来访问scala的ConsumerOffsetChecker类,代码如下:

 

[java] view plain copy
 
  1. import kafka.tools.ConsumerOffsetChecker;     
  2. public class hui   
  3. {  
  4.     public static void main(String[] args)    
  5.     {  
  6.         //适用于kafka0.8.2.0  
  7.         String[] arr = new String[]{"--zookeeper=h71:2181,h72:2181,h73:2181","--group=test-consumer-group"};  
  8.         //适用于kafka0.8.1  
  9. //      String[] arr = new String[]{"--zkconnect=h71:2181,h72:2181,h73:2181","--group=test-consumer-group"};  
  10.         ConsumerOffsetChecker.main(arr);   
  11.     }  
  12. }  

在myeclipse中运行该代码结果为:

 


跟通过kafa-run-class.sh执行的结果是一样的

 

注意:

在http://jxauwxj.iteye.com/blog/2233925博客中所使用的kafka版本是kafka0.8.1,我一开始按其中的方法在Java代码中写入String[] arr = new String[]{"--zookeeper=h71:2181,h72:2181,h73:2181","--group=test-consumer-group"};总是报错:

后来才发现,我用的是kafka0.8.2.0版本,并且这两个版本这块有差异,我查看了源码kafka-0.8.2.0-src.tar\kafka-0.8.2.0-src\core\src\main\scala\kafka\tools\ConsumerOffsetChecker.scala和kafka-0.8.1-src.tar\kafka-0.8.1-src\core\src\main\scala\kafka\tools\ConsumerOffsetChecker.scala

kafka-0.8.2.0:

kafka-0.8.1:

尼玛,这不是坑爹吗。。。。。。

 

并且:在kafka0.8.2.0版本下在Java代码中不能写入"--zookeeper h71:2181,h72:2181,h73:2181","--group test-consumer-group"(在命令行模式下就可以),还必须得有等于号"--zookeeper=h71:2181,h72:2181,h73:2181","--group=test-consumer-group",不加会报错:

分享到:
评论

相关推荐

    kafka-eagle-bin-2.1.0.tar.gz

    kafka-eagle-bin-2.1.0.tar.gz 2022年7月份下载,最新版

    kafka-schema-registry-client-6.2.2.jar

    mvn install:install-file -DgroupId=io.confluent -DartifactId=kafka-schema-registry-client -Dversion=6.2.2 -Dfile=/root/kafka-schema-registry-client-6.2.2.jar -Dpackaging=jar 官网下载地址 packages....

    kafka-eagle-bin-2.0.1.tar.gz

    大数据监控工具kafka监控工具kafka-eagle-bin-1.4.2.tar.gz,比较简单好用。

    kafka-clients-2.4.1-API文档-中英对照版.zip

    赠送jar包:kafka-clients-2.4.1.jar; 赠送原API文档:kafka-clients-2.4.1-javadoc.jar; 赠送源代码:kafka-clients-2.4.1-sources.jar; 赠送Maven依赖信息文件:kafka-clients-2.4.1.pom; 包含翻译后的API文档...

    kafka-clients-2.4.1-API文档-中文版.zip

    赠送jar包:kafka-clients-2.4.1.jar; 赠送原API文档:kafka-clients-2.4.1-javadoc.jar; 赠送源代码:kafka-clients-2.4.1-sources.jar; 赠送Maven依赖信息文件:kafka-clients-2.4.1.pom; 包含翻译后的API文档...

    kafka-eagle-2.0.8.tar.gz

    kafka-eagle-2.0.8.tar.gz

    kafka-eagle-bin-1.4.0.tar.gz

    2019版kafka-eagle-bin-1.4.0 ,附安装连接https://blog.csdn.net/weixin_43231331/article/details/103140698

    kafka_2.9.2-0.8.2.1.tgz

    Kafka默认开启JVM压缩指针,但只是在64位的HotSpot VM受支持,如果安装了32位的HotSpot VM,需要修改/bin/kafka-run-class.sh文件 KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseCompressedOops -XX:+UseParNewGC...

    kafka-clients-0.10.0.1-API文档-中文版.zip

    赠送jar包:kafka-clients-0.10.0.1.jar; 赠送原API文档:kafka-clients-0.10.0.1-javadoc.jar; 赠送源代码:kafka-clients-0.10.0.1-sources.jar; 赠送Maven依赖信息文件:kafka-clients-0.10.0.1.pom; 包含...

    flink-connector-kafka-base-2.11-1.10.0-API文档-中文版.zip

    赠送jar包:flink-connector-kafka-base_2.11-1.10.0.jar; 赠送原API文档:flink-connector-kafka-base_2.11-1.10.0-javadoc.jar; 赠送源代码:flink-connector-kafka-base_2.11-1.10.0-sources.jar; 赠送Maven...

    kafka-eagle-bin-2.0.5.tar.gz

    Kafka图片化界面

    KAFKA-3.1.1-1.3.1.1.p0.2-el7.parcel

    KAFKA-3.1.1-1.3.1.1.p0.2-el7

    kafka-clients-2.0.0-API文档-中英对照版.zip

    赠送jar包:kafka-clients-2.0.0.jar; 赠送原API文档:kafka-clients-2.0.0-javadoc.jar; 赠送源代码:kafka-clients-2.0.0-sources.jar; 赠送Maven依赖信息文件:kafka-clients-2.0.0.pom; 包含翻译后的API文档...

    kafka-eagle-bin-1.3.2.tar.gz

    已编译的kafka-eagle1.3.2直接解压配置后使用具体配置文件地址https://blog.csdn.net/qq_19524879/article/details/82848797

    flink-connector-kafka-2.12-1.14.3-API文档-中文版.zip

    赠送jar包:flink-connector-kafka_2.12-1.14.3.jar 赠送原API文档:flink-connector-kafka_2.12-1.14.3-javadoc.jar 赠送源代码:flink-connector-kafka_2.12-1.14.3-sources.jar 包含翻译后的API文档:flink-...

    kafka-schema-registry-client-3.2.0.jar

    kafka-schema-registry-client-3.2.0.jar包,亲测可用,在aliyun仓库内找不到,可以下载此jar包来进行手动安装

    KAFKA-3.0.0-1.3.0.0.p0.40-el7

    KAFKA-3.0.0-1.3.0.0.p0.40-el7.parcel KAFKA-3.0.0-1.3.0.0.p0.40-el7.parcel.sha1 manifest.json

    kafka-manager-1.3.3.7.zip

    说明:kafka-manager 自己下载编译速度巨慢,此资源是编译好的 kafka-manager,版本是:kafka-manager-1.3.3.7(适用于较新的版本,kafka版本是kafka_2.11-2.0.1)。 安装配置说明: 1. 里头有个自己写的启动脚本,...

    kafka-clients-0.10.0.1-API文档-中英对照版.zip

    赠送jar包:kafka-clients-0.10.0.1.jar; 赠送原API文档:kafka-clients-0.10.0.1-javadoc.jar; 赠送源代码:kafka-clients-0.10.0.1-sources.jar; 包含翻译后的API文档:kafka-clients-0.10.0.1-javadoc-API...

Global site tag (gtag.js) - Google Analytics