AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

如何使用JMX监控Kafka

2016-12-08 3094
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 使用kafka做消息队列中间件时，为了实时监控其性能时，免不了要使用jmx调取kafka broker的内部数据，不管是自己重新做一个kafka集群的监控系统，还是使用一些开源的产品，比如yahoo的kafka manager, 其都需要使用jmx来监控一些敏感的数据。

使用kafka做消息队列中间件时，为了实时监控其性能时，免不了要使用jmx调取kafka broker的内部数据，不管是自己重新做一个kafka集群的监控系统，还是使用一些开源的产品，比如yahoo的kafka manager, 其都需要使用jmx来监控一些敏感的数据。在kafka官网中 http://kafka.apache.org/082/documentation.html#monitoring 这样说：

Kafka uses Yammer Metrics for metrics reporting in both the server and the client. This can be configured to report stats using pluggable stats reporters to hook up to your monitoring system.
The easiest way to see the available metrics to fire up jconsole and point it at a running kafka client or server; this will all browsing all metrics with JMX.

可见kafka官方也是提倡使用jmx并且提供了jmx的调用给用户以监控kafka.

本博文通过使用jmx调用kafka的几个监测项属性来讲述下如何使用jmx来监控kafka.
有关Jmx的使用可以参考：

在使用jmx之前需要确保kafka开启了jmx监控，kafka启动时要添加JMX_PORT=9999这一项，也就是：

JMX_PORT=9999 bin/kafka-server-start.sh config/server.properties &

博主自行搭建了一个kafka集群，只有两个节点。集群中有一个topic（name=default_channel_kafka_zzh_demo）,分为5个partition(0 1 2 3 4).

这里讨论的kafka版本是0.8.1.x和0.8.2.x，这两者在使用jmx监控时会有差异，差异体现在ObjectName之中。熟悉kafka的同学知道，kafka有topic和partition这两个概念，topic中根据一定的策略来分为若干个partitions, 这里就以此举例来看,
在0.8.1.x中有关此项的属性的ObjectName(String值)为：
“kafka.log”:type=”Log”,name=”default_channel_kafka_zzh_demo-*-LogEndOffset”

而在0.8.2.x中有关的属性的ObjectName为：
kafka.log:type=Log,name=LogEndOffset,topic=default_channel_kafka_zzh_demo,partition=0

所以在程序中要区别对待。

这里采用三个监测项来演示如果使用jmx进行监控：

  1. 上面所说的offset (集群中的一个topic下的所有partition的LogEndOffset值，即logSize)
  2. sendCount(集群中的一个topic下的发送总量，这个值是集群中每个broker中此topic的发送量之和)
  3. sendTps(集群中的一个topic下的TPS, 这个值也是集群中每个broker中此topic的发送量之和)

首先是针对单个kafka broker的。

package kafka.jmx;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.management.*;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.io.IOException;
import java.net.MalformedURLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * Created by hidden on 2016/12/8.
 */
public class JmxConnection {
    private static Logger log = LoggerFactory.getLogger(JmxConnection.class);

    private MBeanServerConnection conn;
    private String jmxURL;
    private String ipAndPort = "localhost:9999";
    private int port = 9999;
    private boolean newKafkaVersion = false;

    public JmxConnection(Boolean newKafkaVersion, String ipAndPort){
        this.newKafkaVersion = newKafkaVersion;
        this.ipAndPort = ipAndPort;
    }

    public boolean init(){
        jmxURL = "service:jmx:rmi:///jndi/rmi://" +ipAndPort+ "/jmxrmi";
        log.info("init jmx, jmxUrl: {}, and begin to connect it",jmxURL);
        try {
            JMXServiceURL serviceURL = new JMXServiceURL(jmxURL);
            JMXConnector connector = JMXConnectorFactory.connect(serviceURL,null);
            conn = connector.getMBeanServerConnection();
            if(conn == null){
               log.error("get connection return null!");
                return  false;
            }
        } catch (MalformedURLException e) {
            e.printStackTrace();
            return false;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    public String getTopicName(String topicName){
        String s;
        if (newKafkaVersion) {
            s = "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=" + topicName;
        } else {
            s = "\"kafka.server\":type=\"BrokerTopicMetrics\",name=\"" + topicName + "-MessagesInPerSec\"";
        }
        return s;
    }

    /**
     * @param topicName: topic name, default_channel_kafka_zzh_demo
     * @return 获取发送量(单个broker的，要计算某个topic的总的发送量就要计算集群中每一个broker之和)
     */
public long getMsgInCountPerSec(String topicName){
    String objectName = getTopicName(topicName);
    Object val = getAttribute(objectName,"Count");
    String debugInfo = "jmxUrl:"+jmxURL+",objectName="+objectName;
    if(val !=null){
        log.info("{}, Count:{}",debugInfo,(long)val);
        return (long)val;
    }
    return 0;
}

    /**
     * @param topicName: topic name, default_channel_kafka_zzh_demo
     * @return 获取发送的tps，和发送量一样如果要计算某个topic的发送量就需要计算集群中每一个broker中此topic的tps之和。
     */
    public double getMsgInTpsPerSec(String topicName){
        String objectName = getTopicName(topicName);
        Object val = getAttribute(objectName,"OneMinuteRate");
        if(val !=null){
            double dVal = ((Double)val).doubleValue();
            return dVal;
        }
        return 0;
    }

    private Object getAttribute(String objName, String objAttr)
    {
        ObjectName objectName =null;
        try {
            objectName = new ObjectName(objName);
        } catch (MalformedObjectNameException e) {
            e.printStackTrace();
            return null;
        }
        return getAttribute(objectName,objAttr);
    }

    private Object getAttribute(ObjectName objName, String objAttr){
        if(conn== null)
        {
            log.error("jmx connection is null");
            return null;
        }

        try {
            return conn.getAttribute(objName,objAttr);
        } catch (MBeanException e) {
            e.printStackTrace();
            return null;
        } catch (AttributeNotFoundException e) {
            e.printStackTrace();
            return null;
        } catch (InstanceNotFoundException e) {
            e.printStackTrace();
            return null;
        } catch (ReflectionException e) {
            e.printStackTrace();
            return null;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * @param topicName
     * @return 获取topicName中每个partition所对应的logSize(即offset)
     */
    public Map<Integer,Long> getTopicEndOffset(String topicName){
        Set<ObjectName> objs = getEndOffsetObjects(topicName);
        if(objs == null){
            return null;
        }
        Map<Integer, Long> map = new HashMap<>();
        for(ObjectName objName:objs){
            int partId = getParId(objName);
            Object val = getAttribute(objName,"Value");
            if(val !=null){
                map.put(partId,(Long)val);
            }
        }
        return map;
    }

    private int getParId(ObjectName objName){
        if(newKafkaVersion){
            String s = objName.getKeyProperty("partition");
            return Integer.parseInt(s);
        }else {
            String s = objName.getKeyProperty("name");

            int to = s.lastIndexOf("-LogEndOffset");
            String s1 = s.substring(0, to);
            int from = s1.lastIndexOf("-") + 1;

            String ss = s.substring(from, to);
            return Integer.parseInt(ss);
        }
    }

    private Set<ObjectName> getEndOffsetObjects(String topicName){
        String objectName;
        if (newKafkaVersion) {
            objectName = "kafka.log:type=Log,name=LogEndOffset,topic="+topicName+",partition=*";
        }else{
            objectName = "\"kafka.log\":type=\"Log\",name=\"" + topicName + "-*-LogEndOffset\"";
        }
        ObjectName objName = null;
        Set<ObjectName> objectNames = null;
        try {
            objName = new ObjectName(objectName);
            objectNames = conn.queryNames(objName,null);
        } catch (MalformedObjectNameException e) {
            e.printStackTrace();
            return  null;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }

        return objectNames;
    }
}

注意代码中对于两种不同kafka版本的区别处理。对应前面所说的三个检测项的方法为：

public Map<Integer,Long> getTopicEndOffset(String topicName)
public long getMsgInCountPerSec(String topicName)
public double getMsgInTpsPerSec(String topicName)

对于整个集群的处理需要另外一个类来保证，总体上是对集群中的每一个broker相应的值进行累加，且看代码：

package kafka.jmx;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by hidden on 2016/12/8.
 */
public class JmxMgr {
    private static Logger log = LoggerFactory.getLogger(JmxMgr.class);
    private static List<JmxConnection> conns = new ArrayList<>();

    public static boolean init(List<String> ipPortList, boolean newKafkaVersion){
        for(String ipPort:ipPortList){
            log.info("init jmxConnection [{}]",ipPort);
            JmxConnection conn = new JmxConnection(newKafkaVersion, ipPort);
            boolean bRet = conn.init();
            if(!bRet){
                log.error("init jmxConnection error");
                return false;
            }
            conns.add(conn);
        }
        return true;
    }

    public static long getMsgInCountPerSec(String topicName){
        long val = 0;
        for(JmxConnection conn:conns){
            long temp = conn.getMsgInCountPerSec(topicName);
            val += temp;
        }
        return val;
    }

    public static double getMsgInTpsPerSec(String topicName){
        double val = 0;
        for(JmxConnection conn:conns){
            double temp = conn.getMsgInTpsPerSec(topicName);
            val += temp;
        }
        return val;
    }

    public static Map<Integer, Long> getEndOffset(String topicName){
        Map<Integer,Long> map = new HashMap<>();
        for(JmxConnection conn:conns){
            Map<Integer,Long> tmp = conn.getTopicEndOffset(topicName);
            if(tmp == null){
                log.warn("get topic endoffset return null, topic {}", topicName);
                continue;
            }
            for(Integer parId:tmp.keySet()){//change if bigger
                if(!map.containsKey(parId) || (map.containsKey(parId) && (tmp.get(parId)>map.get(parId))) ){
                    map.put(parId, tmp.get(parId));
                }
            }
        }
        return map;
    }

    public static void main(String[] args) {
        List<String> ipPortList = new ArrayList<>();
        ipPortList.add("10.101.130.1:9999");
        ipPortList.add("10.101.130.2:9999");
        JmxMgr.init(ipPortList,true);

        String topicName = "default_channel_kafka_zzh_demo";
        System.out.println(getMsgInCountPerSec(topicName));
        System.out.println(getMsgInTpsPerSec(topicName));
        System.out.println(getEndOffset(topicName));
    }
}

运行结果：

2016-12-08 19:25:32 -[INFO] - [init jmxConnection [10.101.130.1:9999]] - [kafka.jmx.JmxMgr:20]
2016-12-08 19:25:32 -[INFO] - [init jmx, jmxUrl: service:jmx:rmi:///jndi/rmi://10.101.130.1:9999/jmxrmi, and begin to connect it] - [kafka.jmx.JmxConnection:35]
2016-12-08 19:25:33 -[INFO] - [init jmxConnection [10.101.130.2:9999]] - [kafka.jmx.JmxMgr:20]
2016-12-08 19:25:33 -[INFO] - [init jmx, jmxUrl: service:jmx:rmi:///jndi/rmi://10.101.130.2:9999/jmxrmi, and begin to connect it] - [kafka.jmx.JmxConnection:35]
2016-12-08 20:45:15 -[INFO] - [jmxUrl:service:jmx:rmi:///jndi/rmi://10.101.130.1:9999/jmxrmi,objectName=kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=default_channel_kafka_zzh_demo, Count:6000] - [kafka.jmx.JmxConnection:73]
2016-12-08 20:45:15 -[INFO] - [jmxUrl:service:jmx:rmi:///jndi/rmi://10.101.130.2:9999/jmxrmi,objectName=kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=default_channel_kafka_zzh_demo, Count:4384] - [kafka.jmx.JmxConnection:73]
10384
3.915592283987704E-65
{0=2072, 1=2084, 2=2073, 3=2083, 4=2072}

观察运行结果可以发现 6000+4384 = 10384 = 2072+2084+2073+2083+2072，小伙伴们可以揣摩下原因。
可以通过jconsole连接service:jmx:rmi:///jndi/rmi://10.101.130.1:9999/jmxrmi或者service:jmx:rmi:///jndi/rmi://10.101.130.2:9999/jmxrmi来查看相应的数据值。如下图：
这里写图片描述

也可以通过命令行的形式来查看某项数据，不过这里要借助一个jar包：cmdline-jmxclient-0.10.3.jar，这个请自行下载，网上很多。
将这个jar放入某一目录，博主这里放在了linux系统下的/root/util目录中，以offset举例：
0.8.1.x版-读取topic=default_channel_kafka_zzh_demo,partition=0的Value值：

java -jar cmdline-jmxclient-0.10.3.jar - 10.101.130.1:9999 '"kafka.log":type="Log",name="default_channel_kafka_zzh_demo-0-LogEndOffset"' Value

0.8.2.x版-读取topic=default_channel_kafka_zzh_demo,partition=0的Value值：

java -jar cmdline-jmxclient-0.10.3.jar - 10.101.130.1:9999 kafka.log:type=Log,name=LogEndOffset,topic=default_channel_kafka_zzh_demo,partition=0

看出规律了嘛？如果还是没有，博主再提示一个小技巧，你可以用Jconsole打开相应的属性，然后将鼠标浮于其上，Jconsole会跳出tooltips来提示怎么拼这些属性的ObjectName.

文章标签：
云消息队列 Kafka 版
监控
消息中间件
Kafka
Java
Linux
Apache
中间件
关键词：
云消息队列 Kafka 版监控
监控云消息队列 Kafka 版
朱小厮
目录
相关文章
武子康
|
3月前
|
消息中间件 监控 数据可视化
大数据-79 Kafka 集群模式 集群监控方案 JavaAPI获取集群指标 可视化监控集群方案： jconsole、Kafka Eagle
大数据-79 Kafka 集群模式 集群监控方案 JavaAPI获取集群指标 可视化监控集群方案： jconsole、Kafka Eagle
武子康
161 2
番茄酱脑袋
|
5月前
|
消息中间件 监控 Java
联通实时计算平台问题之监控Kafka集群的断传和积压情况要如何操作
联通实时计算平台问题之监控Kafka集群的断传和积压情况要如何操作
番茄酱脑袋
72 1
AutoMQ
|
7月前
|
消息中间件 监控 Java
「布道师系列文章」宝兰德徐清康解析 Kafka 和 AutoMQ 的监控
本文由北京宝兰德公司解决方案总监徐清康撰写，探讨了Kafka和AutoMQ集群的监控。
AutoMQ
258 2
「布道师系列文章」宝兰德徐清康解析 Kafka 和 AutoMQ 的监控
码农阿豪
|
6月前
|
消息中间件 监控 Java
使用 JMX 监控 Kafka 集群性能指标
使用 JMX 监控 Kafka 集群性能指标
码农阿豪
682 1
欢喜躲在眉梢里
|
6月前
|
消息中间件 NoSQL Kafka
基于Kafka的nginx日志收集分析与监控平台（3）
基于Kafka的nginx日志收集分析与监控平台（3）
欢喜躲在眉梢里
76 0
欢喜躲在眉梢里
|
6月前
|
消息中间件 监控 Kafka
基于Kafka的nginx日志收集分析与监控平台（2）
基于Kafka的nginx日志收集分析与监控平台（2）
欢喜躲在眉梢里
81 0
欢喜躲在眉梢里
|
6月前
|
消息中间件 负载均衡 应用服务中间件
基于Kafka的nginx日志收集分析与监控平台（1）
基于Kafka的nginx日志收集分析与监控平台（1）
欢喜躲在眉梢里
109 0
小王老师呀
|
3月前
|
消息中间件 存储 运维
为什么说Kafka还不是完美的实时数据通道
【10月更文挑战第19天】Kafka 虽然作为数据通道被广泛应用，但在实时性、数据一致性、性能及管理方面存在局限。数据延迟受消息堆积和分区再平衡影响；数据一致性难以达到恰好一次；性能瓶颈在于网络和磁盘I/O；管理复杂性涉及集群配置与版本升级。
小王老师呀
144 1
武子康
|
3月前
|
消息中间件 Java Kafka
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
武子康
68 1
土木林森
|
5月前
|
消息中间件 Java Kafka
Kafka不重复消费的终极秘籍！解锁幂等性、偏移量、去重神器，让你的数据流稳如老狗，告别数据混乱时代！
【8月更文挑战第24天】Apache Kafka作为一款领先的分布式流处理平台，凭借其卓越的高吞吐量与低延迟特性，在大数据处理领域中占据重要地位。然而，在利用Kafka进行数据处理时，如何有效避免重复消费成为众多开发者关注的焦点。本文深入探讨了Kafka中可能出现重复消费的原因，并提出了四种实用的解决方案：利用消息偏移量手动控制消费进度；启用幂等性生产者确保消息不被重复发送；在消费者端实施去重机制；以及借助Kafka的事务支持实现精确的一次性处理。通过这些方法，开发者可根据不同的应用场景灵活选择最适合的策略，从而保障数据处理的准确性和一致性。
土木林森
400 9

热门文章

最新文章

  • 1
    【手把手教你Linux环境下快速搭建Kafka集群】内含脚本分发教程，实现一键部署多个Kafka节点
  • 2
    【Azure Event Hub】Kafka消息发送失败(Timeout Exception)
  • 3
    2024消息队列“四大天王”：Rabbit、Rocket、Kafka、Pulsar巅峰对决
  • 4
    道旅科技借助云消息队列 Kafka 版加速旅游大数据创新发展
  • 5
    2024最全Kafka集群方案汇总
  • 6
    招行面试：RocketMQ、Kafka、RabbitMQ，如何选型？
  • 7
    云消息队列 Kafka 版 V3 系列荣获信通院“云原生技术创新标杆案例”
  • 8
    解决kafka集群由于默认的__consumer_offsets这个topic的默认的副本数为1而存在的单点故障问题
  • 9
    SQLServer CDC数据通过Kafka connect实时同步至分析型数据库 AnalyticDB For PostgreSQL及OSS
  • 10
    springboot配置kafka生产者和消费者详解
  • 1
    Kafka消费组重新平衡流程
    77
  • 2
    Kafka的ISR和OSR的作用分别是什么
    275
  • 3
    Kafka的producer如何实现幂等性
    309
  • 4
    实时计算 Flink版操作报错合集之使用kafka connector时，报错：java.lang.ClassNotFoundException，是什么原因
    526
  • 5
    实时计算 Flink版产品使用问题之建了一张upsert-kafka的flink表，但是数据为空，该如何排查
    79
  • 6
    实时计算 Flink版产品使用问题之在消费Kafka的Avro消息，如何配置FlinkKafka消费者的相关参数
    123
  • 7
    实时计算 Flink版产品使用问题之如何实现两个阿里云账号下的Kafka进行数据的互相传输
    65
  • 8
    实时计算 Flink版产品使用问题之配置了最大连续失败数不为1，在Kafka的精准一次sink中，如果ck失败了，这批数据是否会丢失
    63
  • 9
    AutoMQ vs Kafka: 来自小红书的独立深度评测与对比
    126
  • 10
    AutoMQ 产品动态 | 发布 1.1.0，兼容至 Apache Kafka 3.7，支持 Kaf
    82

    • 相关课程

    更多
  • 消息队列Kafka入门课程
  • 分布式消息系统 Kafka 快速入门
  • 分布式协调系统 Zookeeper 快速入门

    • 相关电子书

    更多
  • Java Spring Boot开发实战系列课程【第16讲】：Spring Boot 2.0 实战Apache Kafka百万级高并发消息中间件与原理解析
  • MaxCompute技术公开课第四季 之 如何将Kafka数据同步至MaxCompute
  • 消息队列kafka介绍

    • 相关实验场景

    更多
  • RocketMQ监控/告警一站式搭建应用
    • 下一篇
    巧用对象存储回源绕过SSRF限制