用Kafka自带脚本实现Kafka消费情况实时监控

查看consumer组内消费的offset

在${KAFKA_HOME}/bin下
./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper host:2181 --group [消费者组名] --topic [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


[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

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); 
    }
}

运行结果如下：

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