kafka获取性能监控数据

1、背景

kafka开源本身实现了yammer metrics，详见https://docs.confluent.io/platform/current/kafka/monitoring.html。只要能获取到metrics的数据kafka对于维护者来说就不再是个黑盒子，每秒进出流量、broker健康度等都可以可视化。

2、通过jconsole查看kafka的mbean

虽然正式部署到生产环境一般不会使用jconsole，但是这一步对于调试非常有用。
本地起一个zk，方便调试我直接通过idea启动kafka，jconsole中找到kafka的进程

查看mbean，个人理解这就是kafka的metrics能提供的所有监控数据，一部分是jdk底层提供的，一部分是kafka自己的。

比如打开java.lang.Thread-》ThreadCount可以看到总线程数

比如打开kafka.controller.ActiveControllerCount，可以看到它的属性和值，该值表示活跃controller个数。

3、通过java客户端获取

首先我们需要给jmx固定一个端口，修改kafka进程的jvm参数，增加
-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=9999
将jmx端口固定为9999，再启动kafka进程

java代码通过“service:jmx:rmi:///jndi/rmi://127.0.0.1:9999/jmxrmi”获取jmx连接

    public static void main(String[] args) {
        String ipAndPort = "127.0.0.1:9999";
        int interval = DEFAULT_INTERVAL_MS;
        String jmxUrl = "service:jmx:rmi:///jndi/rmi://" + ipAndPort + "/jmxrmi";
        boolean isConnected = false;
        while (!isConnected) {
            try {
                JMXServiceURL serviceUrl = new JMXServiceURL(jmxUrl);
                JMXConnector connector = JMXConnectorFactory.connect(serviceUrl, null);
                conn = connector.getMBeanServerConnection();
                isConnected = true;
            } catch (Exception e) {
                LOG.error("start metrics error", e);
            }
            if (!isConnected) {
                try {
                    Thread.sleep(interval);
                } catch (InterruptedException e) {
                    LOG.error("sleep error", e);
                }
            }
        }
  }

使用
AttributeList list = conn.getAttributes(new ObjectName("java.lang:type=Threading"), new String[]{"ThreadCount"});
尝试获取线程数

其他的数据也是一样的获取方式。