笔者经常遇到一些无法优化的慢查询，面对这样的慢查询，笔者会将其进行预先计算存储到mongodb或者elasticsearch中。这个业务场景需要将mysql的binlog数据发送到kafka，然后订阅kafka并消费其中的binlog数据以实现实时加速查询。但是消费binlog就可能会有很多意外发生，比如mysql数据库发生死锁，或者消费发生并发问题，网络长时间阻塞，这些状况都会导致kafka消费发生阻塞，一旦发生阻塞，用户从mongodb或者elasticsearch中就会查询不到最新的mysql数据，所以笔者需要监控kafka中消息的消费情况，监控的方案有很多，笔者进行了整理，以便日后回顾。

kafka-consumer-groups

kafka-consumer-groups.sh是kafka自带的工具，它位于kafka安装目录的bin目录下，它不需要额外下载安装，使用起来非常方便

通过如下命令，我们可以查看kafka的所有消费组

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --list

它的输出如下所示，可以看到其中有一个Test-Group的消费组

我们可以通过如下方式查看某个消费组的消费情况

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group Test-Group

输出结果如下

上述涉及的几个字段意义如下

TOPIC：主题名字

PARTITION：分区id

CURRENT-OFFSET：已经提交的消费位移

LOG-END-OFFSET：总数据量

LAG：未消费的数据量

CONSUMER-ID：代表消费者id，为空代表当前没有消费者

HOST：消费者主机ip，为空代表当前没有消费者

CLIENT-ID：客户端id，为空代表当前没有消费者

kafka-manager

kafka-manager要比kafka-consumer-groups工具强大的多，它不仅可以用于查看消费者消费情况，同时可以管理Topic，包括创建topic，管理partition。

从下面地址下载kafka-manager的安装包

https://github.com/yahoo/kafka-manager/releases

解压并进入kafka安装目录，使用sbt进行编译，sbt需要配置源，否则编译速度非常慢

sbt clean dist

将编译好的zip格式文件复制出来，解压缩，修改conf目录下的application.conf文件，修改zookeeper地址参数kafka-manager.zkhosts，在bin目录下执行./kafka-manager命令，kafka默认端口是9000，进入管理页面之后配置kafka节点相关信息，就能监控kafka运行情况

如下是kafka-manager的管理界面

查看所有的消费组

Broker管理的partition列表

某个消费组消费某个topic的消费状况