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

1.查看consumer组内消费的offset：

./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper **:2181 --group ** --topic **
1

2.其中的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时才会有，关闭消费者进程该组会自动消失
1
2
3
4

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

3.执行结果如下：

  列出了所有消费者组的所有信息，包括Group(消费者组)、Topic、Pid(分区id)、Offset(当前已消费的条数)、LogSize(总条数)、Lag(未消费的条数)、Owner

[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
或者：
[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
再或者：
[hadoop@h71 kafka_2.10-0.8.2.0]$ bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper=h71:2181 --group=WordcountConsumerGroup
Group           Topic                          Pid Offset          logSize         Lag             Owner
test-consumer-group test                           0   4               40              36              none
test-consumer-group test                           1   14              57              43              none

[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
Group           Topic                          Pid Offset          logSize         Lag             Owner
WordcountConsumerGroup test                           0   9               40              31              none
WordcountConsumerGroup test                           1   21              57              36              none
[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
Group           Topic                          Pid Offset          logSize         Lag             Owner
console-consumer-42639 test                           0   40              40              0               console-consumer-42639_h71-1498000434797-f1c703cf-0
console-consumer-42639 test                           1   57              57              0               console-consumer-42639_h71-1498000434797-f1c703cf-0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

注意：Apache的kafka_2.11-0.11.0.2版本可执行如下命令

bin/kafka-consumer-offset-checker.sh --zookeeper 192.166.240.11:2181 --topic weishiDataOGG  --group cb6e251e-b61b-48df-985e-636a54bf8daf
[2020-04-06 16:40:14,092] WARN WARNING: ConsumerOffsetChecker is deprecated and will be dropped in releases following 0.9.0. Use ConsumerGroupCommand instead. (kafka.tools.ConsumerOffsetChecker$)
Group           Topic                          Pid Offset          logSize         Lag             Owner
cb6e251e-b61b-48df-985e-636a54bf8daf weishiDataOGG                  0   42565489        58261852        15696363        none
[oracle@localhost kafka]$ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic weishiDataOGG --time -1 --broker-list 192.166.240.33:9092 --partitions 0
weishiDataOGG:0:58261852
1
2
3
4
5
6

4.java代码执行kafka-run-class.sh：

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

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

在myeclipse中运行该代码结果跟通过kafa-run-class.sh执行的结果是一样的：

注意：
  在kafka监控之kafka-run-class.sh博客中所使用的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"，不加会报错：