`
kavy
  • 浏览: 888680 次
  • 性别: 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_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_2.10-0.8.2.1.tgz

    4. 监控与管理:通过Kafka提供的监控工具,如`kafka-console-consumer.sh`和`kafka-run-class.sh`,监控主题和消费者状态。 六、大数据应用场景 1. 数据管道:Kafka作为实时数据流处理的基础设施,用于收集、存储和...

    kafka集群搭建及测试.docx

    为了确保Kafka集群的稳定运行,你需要定期检查节点状态,使用Kafka提供的命令行工具如`bin/kafka-topics.sh --describe --bootstrap-server localhost:9092`来查看topic详情,或`bin/kafka-run-class.sh kafka....

    Kafka跨集群容灾备份.docx

    $ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9093 --topic test --time -1 ``` 2. 计算两者之差即为当前分区的消息数量。 #### 五、跨机房备份案例 假设有一个简单的场景,...

    kafka管理工具 监控 偏移量 消费 offset,必备技能.docx

    使用 `bin/kafka-consumer-groups.sh` 工具,可以查看消费者组的信息,包括消费者的 Offset、Lag、LogSize 等信息。例如: `bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server hadoop-4:9092,hadoop...

    Kafka 常用命令行详细介绍及整理

    使用 `kafka-run-class.sh` 运行 `kafka.tools.ConsumerOffsetChecker`,查看消费者组在特定 Topic 上的消费进度: ```bash kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper localhost:2181 -...

    Kafka基础操作

    - **查看日志**:使用 `kafka-run-class kafka.tools.DumpLogSegments` 命令可以查看 Kafka 的日志信息。例如,查看存储在 `/tmp/kafka-logs/` 目录下的日志文件: ``` kafka-run-class kafka.tools....

    kafka-ganglia:使用神经节监控Kafka指标的快速而肮脏的方法

    确保编辑kafka-run-class.sh以包含以下内容: KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false " 如果要...

    linux中如何操作kafka

    bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper localhost:3181 --topic topic_open_flow_es_99 ``` 注意:在删除主题前,必须先停止所有的生产者和消费者,否则可能会导致数据丢失。 #### 四...

    Kafka性能测试报告.pdf

    使用Kafka提供的异步API性能测试工具kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance来进行测试。测试结果表明,异步API的性能非常高,对CPU进行监控,发现CPU的利用率只是在10%-20%之间,...

    windows搭建kafka环境

    3. 修改kafka-run-class.bat文件 此文件的修改是将相对路径进行调整,例如: - 将“pushd%~dp0..\..”修改为“pushd%~dp0”。 - 将“%BASE_DIR%/config”修改为“%BASE_DIR%/../config”。 - 需要调整的类路径信息是...

    利用docker和docker-compose部署单机kafka

    - 启动消费者:`kafka-console-consumer.sh --bootstrap-server 192.168.1.60:9094 --topic kafeidou --from-beginning` - 启动生产者:`kafka-console-producer.sh --broker-list 192.168.1.60:9092 --topic ...

    kafka1:ff

    Kafka 提供了一系列的命令行工具,如 `kafka-topics.sh` 用于创建、查询和修改 topics,`kafka-console-consumer.sh` 和 `kafka-console-producer.sh` 分别用于消费和生产消息,以及 `kafka-run-class.sh` 用于运行 ...

    kafka-test:Apache kafka 测试

    1. **本地测试集群**: 使用像`kafka-run-class.sh`这样的工具,可以在本地启动一个小型的Kafka集群,这样就可以进行真实的生产和消费操作。在测试中,你可以发送和接收消息,验证它们是否正确传输。 2. **使用Kafka...

    fink_java_

    4. Connectors:Flink提供了多种数据源和接收器,如Kafka、HDFS、Cassandra等,便于与其他系统集成。 通过以上的学习,你应该对Flink的基本安装、Java API的使用以及基本操作有了初步了解。继续深入研究Flink的高级...

    linux_flink-1.13.0.rar

    在Linux环境中,Apache Flink是一个流行的开源大数据处理框架,它提供了高效的流处理和批处理能力。Flink 1.13.0版本是其稳定的一个发行版,包含了多个改进和新特性。本教程将深入讲解如何在Linux系统上将Flink部署...

Global site tag (gtag.js) - Google Analytics