我刚开始用一个简单的例子来阅读jmx指标,并使用HeapMemoryUsage,CPUTime等的简单值来获得它的感觉。我需要尝试访问kafka服务器/消费者指标特别滞后,我可以看到它在FetcherLagMetrics-ConsumerLag下的jconsole应用程序中可见。但programmaticaly我收到以下错误:
javax.management.InstanceNotFoundException: kafka.consumer:type=consumer-
fetch-manager-metrics
这告诉我,消费者获取管理器指标是问题,因为它甚至在jconsole中也不存在。我把它改成了以下但仍然是同一个问题:
consumerBean = jmxCon.getMBeanServerConnection().getAttribute(new
ObjectName("kafka.server:type=FetcherLagMetrics"),"ConsumerLag");
cd = (CompositeData) consumerBean;
尝试访问这些值的代码如下:
jmxCon.getMBeanServerConnection().invoke(new
ObjectName("java.lang:type=Memory"), "gc", null, null);
for (int i = 0; i < 100; i++) {
//get an instance of the HeapMemoryUsage Mbean
memoryMbean = jmxCon.getMBeanServerConnection().getAttribute(new ObjectName("java.lang:type=Memory"), "HeapMemoryUsage");
cd = (CompositeData) memoryMbean;
//get an instance of the OperatingSystem Mbean
osMbean = jmxCon.getMBeanServerConnection().getAttribute(new ObjectName("java.lang:type=OperatingSystem"),"ProcessCpuTime");
//get an instance of the kafka metrics Mbean
consumerBean = jmxCon.getMBeanServerConnection().getAttribute(new ObjectName("kafka.consumer:type=consumer-fetch-manager-metrics"),"MaxLag");
cd = (CompositeData) consumerBean;
consumerBeanII = jmxCon.getMBeanServerConnection().getAttribute(new ObjectName("kafka.server:type=FetcherLagMetrics,name=ConsumerLag"),"Lag");
System.out.println("Used memory: " + " " + cd.get("MaxLag") + " Used cpu: " + consumerBean); //print memory usage
tempMemory = tempMemory + Long.parseLong(cd.get("used").toString());
Thread.sleep(1000); //delay for one second
}
它在线路上失败了consumerBean = ......有人可以解释或提供使用jmx / JMI实现访问kafka指标的正确方法吗?
我不确定你的Kafka版本,但是当我使用JConsole查看我的版本(v1.1.0)时,你所描述的消费者滞后没有这样的bean。因此,我认为您的JMX查询将失败。
相反,我相信这种信息被转移到Kafka Admin API界面。因此,如果您不想使用kafka-consumer-groups.sh
命令行实用程序,则必须立即使用此方法来获取滞后设置。
对我们来说,我们只需要这些统计数据进行监控,因此我们使用https://github.com/danielqsj/kafka_exporter获取Prometheus的信息。