Kafka监控工具KafkaOffsetMonitor配置及使用

淩亂°似流年 2022-05-09 13:56 253阅读 0赞

## 一、KafkaOffsetMonitor简述 ##

KafkaOffsetMonitor是Kafka的一款客户端消费监控工具，用来实时监控Kafka服务的Consumer以及它们所在的Partition中的Offset，我们可以浏览当前的消费者组，并且每个Topic的所有Partition的消费情况都可以一目了然。

## 二、KafkaOffsetMonitor下载 ##

KafkaOffsetMonitor托管在Github上，可以通过Github下载。  
下载地址：https://github.com/quantifind/KafkaOffsetMonitor/releases

或者下载百度网盘：链接：https://pan.baidu.com/s/1geEBEvT 密码：jaeu

![1209537-20180108132411629-753572802.png][]

## 三、KafkaOffsetMonitor启动 ##

将下载下来的KafkaOffsetMonitor jar包上传到linux上，可以新建一个目录KafkaMonitor，用于存放KafkaOffsetMonitor-assembly-0.2.0.jar进入到KafkaMonitor目录下，通过java编译命令来运行这个jar包：

![复制代码][copycode.gif]

[root@kafka50 KafkaMonitor]# java -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk 10.0.0.50:12181,10.0.0.60:12181,10.0.0.70:12181 --port 8088  --refresh 5.seconds --retain 1.days
    按回车后，可以看到控制台输出：
    
    serving resources from: jar:file:/data/KafkaMonitor/KafkaOffsetMonitor-assembly-0.2.0.jar!/offsetapp
    SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
    SLF4J: Defaulting to no-operation (NOP) logger implementation
    SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
    2018-01-05 21:17:36.267:INFO:oejs.Server:jetty-7.x.y-SNAPSHOT
    log4j:WARN No appenders could be found for logger (org.I0Itec.zkclient.ZkConnection).
    log4j:WARN Please initialize the log4j system properly.
    log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
    2018-01-05 21:17:36.630:INFO:oejsh.ContextHandler:started o.e.j.s.ServletContextHandler{/,jar:file:/data/KafkaMonitor/KafkaOffsetMonitor-assembly-0.2.0.jar!/offsetapp}
    2018-01-05 21:17:36.662:INFO:oejs.AbstractConnector:Started SocketConnector@0.0.0.0:8088

![复制代码][copycode.gif]

如果没有指定端口，则默认会开启一个随机端口。

![复制代码][copycode.gif]

参数说明:
    
    zk ：zookeeper主机地址，如果有多个，用逗号隔开
    port ：应用程序端口
    refresh ：应用程序在数据库中刷新和存储点的频率
    retain ：在db中保留多长时间
    dbName ：保存的数据库文件名，默认为offsetapp

![复制代码][copycode.gif]

为了更方便的启动KafkaOffsetMonitor，可以写一个启动脚本来直接运行，我这里新建一个名为：kafka-monitor-start.sh的脚本，然后编辑这个脚本：

![复制代码][copycode.gif]

[root@kafka50 KafkaMonitor]# vim kafka-monitor-start.sh 
    java -Xms512M -Xmx512M -Xss1024K -XX:PermSize=256m -XX:MaxPermSize=512m  -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb \
    --port 8088 \
    --zk 10.0.0.50:12181,10.0.0.60:12181,10.0.0.70:12181 \
    --refresh 5.minutes \
    --retain 1.day >/dev/null 2>&1;

![复制代码][copycode.gif]

然后退出保存即可，接下来修改一下kafka-monitor-start.sh的权限

[root@kafka50 KafkaMonitor]# chmod +x kafka-monitor-start.sh

启动KafkaOffsetMonitor：

[root@kafka50 KafkaMonitor]# nohup /data/KafkaMonitor/kafka-monitor-start.sh &
    [1] 6551
    [root@kafka50 KafkaMonitor]# lsof -i:8088
    COMMAND  PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
    java    6552 root   16u  IPv6  26047      0t0  TCP *:radan-http (LISTEN)

## 四、KafkaOffsetMonitor Web UI ##

在游览器中输入：http://ip:port即可以查看KafkaOffsetMonitor Web UI,如下图：

![1209537-20180108145441926-909126059.png][]

在下图中有一个Visualizations选项卡，点击其中的Cluster Overview可以查看当前Kafka集群的Broker情况

![1209537-20180108145543457-1265044363.png][]

# 五、简单的Producer #

## 1、新建一个Topic ##

首先为本次试验新建一个Topic，命令如下：

![复制代码][copycode.gif]

bin/kafka-topics.sh \
        --create \
        --zookeeper 10.0.0.50:12181 \
        --replication-factor 3 \
        --partition 3 \
        --topic kafkamonitor-simpleproducer

![复制代码][copycode.gif]

## 2、新建SimpleProducer代码 ##

在上一篇文章中提到的Producer封装Github代码的基础上，写了一个往kafkamonitor-simpleproducer发送message的java代码。

![复制代码][copycode.gif]

import com.ckm.kafka.producer.impl.KafkaProducerToolImpl;
    import com.ckm.kafka.producer.inter.KafkaProducerTool;
    
    /**
     * Created by ckm on 2016/8/30.
     */
    public class SimpleProducer {
        public static void main(String[] args) {
            KafkaProducerTool kafkaProducerTool = new KafkaProducerToolImpl();
            int i = 0;
            String message = "";
            while (true) {
                message = "test-simple-producer : " + i ++;
                kafkaProducerTool.publishMessage("kafkamonitor-simpleproducer", message);
            }
        }
    }

![复制代码][copycode.gif]

程序运行效果：   
　　![这里写图片描述][20160830222804561]

## 3、ConsoleConsumer消费该topic ##

用kafka自带的ConsoleConsumer消费kafkamonitor-simpleproducer中的message。

bin/kafka-console-consumer.sh --zookeeper m000:2181 --from-beginning --topic kafkamonitor-simpleproducer

消费截图如下：   
　　![这里写图片描述][20160830222846578]

## 4、KafkaOffsetMonitor页面 ##

（1）在Topic List选项卡中，我们可以看到刚才新建的`kafkamonitor-simpleproducer`   
　　![这里写图片描述][20160830222500478]  
（2）点开后，能看到有一个console-consumer正在消费该topic   
　　![这里写图片描述][20160830222545417]  
（3）继续进入该Consumer，可以查看该Consumer当前的消费状况   
　　![这里写图片描述][20160830223625494]  
　　这张图片的左上角显示了当前Topic的生产速率，右上角显示了当前Consumer的消费速率。   
　　图片中还有三种颜色的线条，蓝色的表示当前Topic中的Message数目，灰色的表示当前Consumer消费的offset位置，红色的表示蓝色灰色的差值，即当前Consumer滞后于Producer的message数目。   
（4）看一眼各partition中的message消费情况   
　　![这里写图片描述][20160830223410169]  
　　从上图可以看到，当前有3个Partition，每个Partition中的message数目分布很不均匀。这里可以与接下来的自定义Producer的情况进行一个对比。

# 六、自定义Partitioner的Producer #

## 1、新建一个Topic ##

![复制代码][copycode.gif]

bin/kafka-topics.sh \
        --create \
        --zookeeper 10.0.0.50:12181 \
        --replication-factor 3 \
        --partition 3 \
        --topic kafkamonitor-partitionedproducer

![复制代码][copycode.gif]

## *2、Partitioner代码* ##

逻辑很简单，循环依次往各Partition中发送message。

![复制代码][copycode.gif]

import kafka.producer.Partitioner;
    
    /**
     * Created by ckm on 2018/1/8.
     */
    public class TestPartitioner implements Partitioner {
        public TestPartitioner() {
        }
    
        @Override
        public int partition(Object key, int numPartitions) {
            int intKey = (int) key;
            return intKey % numPartitions;
        }
    }

![复制代码][copycode.gif]

## 3、Producer代码 ##

将自定义的Partitioner设置到Producer，其他调用过程和二中类似。

![复制代码][copycode.gif]

import com.ckm.kafka.producer.impl.KafkaProducerToolImpl;
    import com.ckm.kafka.producer.inter.KafkaProducerTool;
    
    /**
     * Created by ckm on 2016/8/30.
     */
    public class PartitionedProducer {
        public static void main(String[] args) {
            KafkaProducerTool kafkaProducerTool = new KafkaProducerToolImpl();
            kafkaProducerTool.getProducerProperties().put("partitioner.class", "TestPartitioner");
            int i = 0;
            String message = "";
            while (true) {
                message = "test-partitioner-producer : " + i;
                System.out.println(message);
                kafkaProducerTool.publishPartitionedMessage("kafkamonitor-partitionedproducer", i + "", message);
                i ++;
            }
        }
    }

![复制代码][copycode.gif]

代码运行效果如下图：   
　　![这里写图片描述][20160830224559116]

## 4、ConsoleConsumer消费Message ##

bin/kafka-console-consumer.sh --zookeeper 10.0.0.50:12181 --from-beginning --topic kafkamonitor-partitionedproducer

消费效果如下图：   
　　![这里写图片描述][20160830224727462]

## 5、KafkaOffsetMonitor页面 ##

其他页面与上面的类似，这里只观察一下每个partition中的message数目与第二节中的对比。可以看到这里每个Partition中message分别是很均匀的。   
　　![这里写图片描述][20160830225015747]

**注意事项：**   
　　注意这里有一个坑，默认情况下Producer往一个不存在的Topic发送message时会自动创建这个Topic。由于在这个封装中，有同时传递message和topic的情况，如果调用方法时传入的参数反了，将会在Kafka集群中自动创建Topic。在正常情况下，应该是先把Topic根据需要创建好，然后Producer往该Topic发送Message，最好把Kafka这个默认自动创建Topic的功能关掉。   
　　那么，假设真的不小心创建了多余的Topic，在删除时，会出现“marked for deletion”提示，只是将该topic标记为删除，使用list命令仍然能看到。如果需要调整这两个功能的话，在server.properties中配置如下两个参数：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>默认值</th> 
   <th>作用</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>auto.create.topics.enable</td> 
   <td>true</td> 
   <td>Enable auto creation of topic on the server</td> 
  </tr> 
  <tr> 
   <td>delete.topic.enable</td> 
   <td>false</td> 
   <td>Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off</td> 
  </tr> 
 </tbody> 
</table>

# #

## 七，KafkaOffsetMonitor 总结 ##

KafkaOffsetMonitor：程序一个jar包的形式运行，部署较为方便。只有监控功能，使用起来也较为安全。  
除了KafkaOffsetMonitor，Kafka监控工具还有另外两款：  
Kafka Web Console：监控功能较为全面，可以预览消息，监控Offset、Lag等信息，但存在bug，不建议在生产环境中使用。  
Kafka Manager：偏向Kafka集群管理，若操作不当，容易导致集群出现故障。对Kafka实时生产和消费消息是通过JMX实现的。没有记录Offset、Lag等信息。

[1209537-20180108132411629-753572802.png]: /images/20220506/3460f6c1b609466daeb992831ad6a3c3.png
[copycode.gif]: /images/20220506/4e5e1f389bf4404da03a0b38b532eda6.png
[1209537-20180108145441926-909126059.png]: /images/20220506/e6dca0c2b90144d181380bf582fbf241.png
[1209537-20180108145543457-1265044363.png]: /images/20220506/c157fc7109874c4caece664f20d73b35.png
[20160830222804561]: /images/20220506/bd99160020874939ae4e2d90c95a8ef5.png
[20160830222846578]: /images/20220506/ba1dd8e9e84d409faa6347842dd31b9f.png
[20160830222500478]: /images/20220506/9515f9a00ace4d3d87cb7fc75788cfb9.png
[20160830222545417]: /images/20220506/c2c21b25a4334b828191c132d24a9c37.png
[20160830223625494]: /images/20220506/254f250ad27245669977ae6e1aec052f.png
[20160830223410169]: /images/20220506/d1d9c0072f8c469fa3f6553146c46d62.png
[20160830224559116]: /images/20220506/333ca49dac344dd28f80ae6ee524a271.png
[20160830224727462]: /images/20220506/3ce9d19143444395ba15b9fad8ac28eb.png
[20160830225015747]: /images/20220506/2a4929fcba7c4f3b9a8fc6972e00485c.png