通过rmi来调用kafka里的jmx信息

 kafka的官方文档上，已经列出了所有的mxbean,链接：http://kafka.apache.org/documentation.html#monitoring

通过jconsole也能看到。

 

 

 

package com.wxj.kafka.monitor.jmx;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;

import javax.management.JMX;
import javax.management.MBeanServerConnection;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectInstance;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;

import com.yammer.metrics.reporting.JmxReporter;

/**
 * 通过rmi来调用kafka里的jmx信息
 * @author root
 *
 */
public class JMX1
{

	public static void main(String[] args)
	{
		try
		{
			String rmiAddress = "192.168.199.131:9999";
			HashMap<String, Object> prop = new HashMap<String, Object>();
			JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://" + rmiAddress + "/jmxrmi");
			JMXConnector conn = JMXConnectorFactory.connect(url, prop);
			MBeanServerConnection mbsc = conn.getMBeanServerConnection();
			
			
			printAllTopicsBytesInPerSec(mbsc);
			
			//printAllTopicsBytesOutPerSec(mbsc);
			
			//printMBeans(mbsc);

		} catch (Exception e)
		{
			e.printStackTrace();
		}

	}

	private static void printAllTopicsBytesOutPerSec(MBeanServerConnection mbsc) throws MalformedObjectNameException
	{
		String name = "\"kafka.server\":type=\"BrokerTopicMetrics\",name=\"AllTopicsBytesOutPerSec\"";
		ObjectName mbeanName = new ObjectName(name);
		
		JmxReporter.MeterMBean meterMBean = JMX.newMBeanProxy(mbsc, mbeanName, JmxReporter.MeterMBean.class);
		System.out.println(meterMBean.getCount() + "," + meterMBean.getEventType() + "," + meterMBean.getFifteenMinuteRate()
				+ meterMBean.getFiveMinuteRate() + "," + meterMBean.getMeanRate() + "," + meterMBean.getOneMinuteRate() + 
				"," + meterMBean.getRateUnit());
	}

	private static void printAllTopicsBytesInPerSec(MBeanServerConnection mbsc) throws MalformedObjectNameException
	{
		String name = "\"kafka.server\":type=\"BrokerTopicMetrics\",name=\"AllTopicsBytesInPerSec\"";
		ObjectName mbeanName = new ObjectName(name);
		
		JmxReporter.MeterMBean meterMBean = JMX.newMBeanProxy(mbsc, mbeanName, JmxReporter.MeterMBean.class);
		System.out.println(meterMBean.getCount() + "," + meterMBean.getEventType() + "," + meterMBean.getFifteenMinuteRate()
				+ meterMBean.getFiveMinuteRate() + "," + meterMBean.getMeanRate() + "," + meterMBean.getOneMinuteRate() + 
				"," + meterMBean.getRateUnit());
	}

	/**
	 * 打印所有的mxbean
	 * @param mbsc
	 * @throws IOException
	 */
	private static void printMBeans(MBeanServerConnection mbsc) throws IOException
	{
		Set MBeanset = mbsc.queryMBeans(null, null);
		System.out.println("MBeanset.size() : " + MBeanset.size());
		Iterator MBeansetIterator = MBeanset.iterator();
		while (MBeansetIterator.hasNext())
		{
			ObjectInstance objectInstance = (ObjectInstance) MBeansetIterator.next();
			System.out.println(objectInstance.getObjectName());
		}
	}

}