使用JMX监控Zookeeper状态Java API

一、背景

上一篇通过java自带的jconsole来获取zookeeper状态。主要有几个不方便的地方，zk集群一般会部署3或者5台，在多个jconsole窗口中切换比较麻烦，各个zk服务及历史数据之间，不能直观比较。一般会做一个web管理页面来展示集群状态，设置报警阀值来做报警。

二、jvm平台提供mbeans

在java5.0以上版本，有一组api可以让java应用程序和允许的工具监视和管理java虚拟机（jvm）和虚拟机所在的本机操作系统。该组api在 java.lang.management包。可以通过这些api可以监控local端jvm，同时也可以监控远端jvm。

通过静态工厂方法获取mxbean实例，从本地访问正在运行的虚拟机的mxbean接口。这些bean我们从managementfactory类中定义的静态方法获取；如managementfactory.getoperatingsystemmxbean();其中不足就是只能获取本地的jvm状态。无法获取远程的虚拟机数据。

• classloadingmxbean java虚拟机的类加载系统
• compilationmxbean java虚拟机的编译系统
• memorymxbean java虚拟机的内存系统
• runtimemxbean java虚拟机的运行时系统
• operatingsystemmxbean java虚拟机在其上运行的操作系统
• garbagecollectormxbean java虚拟机中的垃圾回收器
• memorymanagermxbean java虚拟机中的内存管理器
• memorypoolmxbean java虚拟机中的内存池

三、zookeeper提供出来的mbeans

构造mxbean代理实例，通过代理将方法调用转发到给定的mbeanserve。jconsole能够监控的项目，通过api都能获取到。

具体代码如下：

import java.io.ioexception;
import java.lang.management.classloadingmxbean;
import java.lang.management.compilationmxbean;
import java.lang.management.managementfactory;
import java.lang.management.operatingsystemmxbean;
import java.lang.management.threadmxbean;
import java.util.arraylist;
import java.util.arrays;
import java.util.collections;
import java.util.hashmap;
import java.util.iterator;
import java.util.list;
import java.util.map;
import java.util.set;
import javax.management.instancenotfoundexception;
import javax.management.introspectionexception;
import javax.management.jmx;
import javax.management.mbeanserverconnection;
import javax.management.malformedobjectnameexception;
import javax.management.objectname;
import javax.management.reflectionexception;
import javax.management.remote.jmxconnector;
import javax.management.remote.jmxconnectorfactory;
import javax.management.remote.jmxserviceurl;
import org.apache.zookeeper.server.connectionmxbean;
import org.apache.zookeeper.server.datatreemxbean;
import org.apache.zookeeper.server.zookeeperservermxbean;
public class zkjmxtest {
  static jmxconnector connector;
  /**
   * @param args
   * @throws ioexception
   * @throws malformedobjectnameexception
   * @throws instancenotfoundexception
   * @throws reflectionexception
   * @throws introspectionexception
   */
  public static void main(string[] args) throws ioexception, malformedobjectnameexception,
    instancenotfoundexception, introspectionexception, reflectionexception {
    operatingsystemmxbean osbean = managementfactory.getoperatingsystemmxbean();
    system.out.println("体系结构:" + osbean.getarch());//操作系统体系结构
    system.out.println("处理器核数:" + osbean.getavailableprocessors());///核数
    system.out.println("名字:" + osbean.getname());//名字
    system.out.println(osbean.getversion());//操作系统版本
    threadmxbean threadbean=managementfactory.getthreadmxbean();
    system.out.println("活动线程:" + threadbean.getthreadcount());//总线程数
    classloadingmxbean classloadingmxbean = managementfactory.getclassloadingmxbean();
    compilationmxbean compilationmxbean = managementfactory.getcompilationmxbean();
    system.out.println("===========");
    // 通过 mbeanserver间接地访问 mxbean 接口
    mbeanserverconnection mbsc = creatembeanserver("192.168.1.100", "9991", "controlrole", "123456");
    // 操作系统
    objectname os = new objectname("java.lang:type=operatingsystem");
    system.out.println("体系结构:" + getattribute(mbsc, os, "arch"));//体系结构
    system.out.println("处理器核数:" + getattribute(mbsc, os, "availableprocessors"));//核数
    system.out.println("总物理内存:" + getattribute(mbsc, os, "totalphysicalmemorysize"));//总物理内存
    system.out.println("空闲物理内存:" + getattribute(mbsc, os, "freephysicalmemorysize"));//空闲物理内存
    system.out.println("总交换空间:" + getattribute(mbsc, os, "totalswapspacesize"));//总交换空间
    system.out.println("空闲交换空间:" + getattribute(mbsc, os, "freeswapspacesize"));//空闲交换空间
    system.out.println("操作系统:" + getattribute(mbsc, os, "name")+ getattribute(mbsc, os, "version"));//操作系统
    system.out.println("提交的虚拟内存:" + getattribute(mbsc, os, "committedvirtualmemorysize"));//提交的虚拟内存
    system.out.println("系统cpu使用率:" + getattribute(mbsc, os, "systemcpuload"));//系统cpu使用率
    system.out.println("进程cpu使用率:" + getattribute(mbsc, os, "processcpuload"));//进程cpu使用率
    system.out.println("============");//
    // 线程
    objectname threading = new objectname("java.lang:type=threading");
    system.out.println("活动线程:" + getattribute(mbsc, threading, "threadcount"));// 活动线程
    system.out.println("守护程序线程:" + getattribute(mbsc, threading, "daemonthreadcount"));// 守护程序线程
    system.out.println("峰值:" + getattribute(mbsc, threading, "peakthreadcount"));// 峰值
    system.out.println("启动的线程总数:" + getattribute(mbsc, threading, "totalstartedthreadcount"));// 启动的线程总数
    threadmxbean threadbean2 = managementfactory.newplatformmxbeanproxy
        (mbsc, managementfactory.thread_mxbean_name, threadmxbean.class);
    system.out.println("活动线程:" + threadbean2.getthreadcount());// 活动线程
    threadmxbean threadbean3 = managementfactory.getthreadmxbean();
    system.out.println("本地活动线程:" + threadbean3.getthreadcount());// 本地活动线程
    system.out.println("============");//
    objectname compilation = new objectname("java.lang:type=compilation");
    system.out.println("总编译时间 毫秒:" + getattribute(mbsc, compilation, "totalcompilationtime"));// 总编译时间 毫秒
    system.out.println("============");//
    objectname classloading = new objectname("java.lang:type=classloading");
    system.out.println("已加载类总数:" + getattribute(mbsc, classloading, "totalloadedclasscount"));// 已加载类总数
    system.out.println("已加装当前类:" + getattribute(mbsc, classloading, "loadedclasscount"));// 已加装当前类
    system.out.println("已卸载类总数:" + getattribute(mbsc, classloading, "unloadedclasscount"));// 已卸载类总数
    system.out.println("==========================================================");//
    // http://zookeeper.apache.org/doc/r3.4.6/zookeeperjmx.html
    // org.apache.zookeeperservice:name0=replicatedserver_id1,name1=replica.1,name2=follower
    objectname replica = new objectname("org.apache.zookeeperservice:name0=replicatedserver_id1,name1=replica.1");
    system.out.println("replica.1运行状态:" + getattribute(mbsc, replica, "state"));// 运行状态
    mbsc = creatembeanserver("192.168.1.100", "9992", "controlrole", "123456");
    system.out.println("==============节点树对象===========");
    objectname datatreepattern = new objectname("org.apache.zookeeperservice:name0=replicatedserver_id?,name1=replica.?,name2=*,name3=inmemorydatatree");
    set<objectname> datatreesets = mbsc.querynames(datatreepattern, null);
    iterator<objectname> datatreeiterator = datatreesets.iterator();
    // 只有一个
    while (datatreeiterator.hasnext()) {
      objectname datatreeobjectname = datatreeiterator.next();
      datatreemxbean datatree = jmx.newmbeanproxy(mbsc, datatreeobjectname, datatreemxbean.class);
      system.out.println("节点总数:" + datatree.getnodecount());// 节点总数
      system.out.println("watch总数:" + datatree.getwatchcount());// watch总数
      system.out.println("临时节点总数:" + datatree.countephemerals());// watch总数
      system.out.println("节点名及字符总数:" + datatree.approximatedatasize());// 节点全路径和值的总字符数
      map<string, string> datatreemap = datatreeobjectname.getkeypropertylist();
      string replicaid = datatreemap.get("name1").replace("replica.", "");
      string role = datatreemap.get("name2");// follower,leader,observer,standalone
      string canonicalname = datatreeobjectname.getcanonicalname();
      int roleendindex = canonicalname.indexof(",name3");
      objectname roleobjectname = new objectname(canonicalname.substring(0, roleendindex));
      system.out.println("==============zk服务状态===========");
      zookeeperservermxbean zookeeperserver = jmx.newmbeanproxy(mbsc, roleobjectname, zookeeperservermxbean.class);
      system.out.println(role + " 的ip和端口:" + zookeeperserver.getclientport());// ip和端口
      system.out.println(role + " 活着的连接数:" + zookeeperserver.getnumaliveconnections());// 连接数
      system.out.println(role + " 未完成请求数:" + zookeeperserver.getoutstandingrequests());// 未完成的请求数
      system.out.println(role + " 接收的包:" + zookeeperserver.getpacketsreceived());// 收到的包
      system.out.println(role + " 发送的包:" + zookeeperserver.getpacketssent());// 发送的包
      system.out.println(role + " 平均延迟（毫秒）:" + zookeeperserver.getavgrequestlatency());
      system.out.println(role + " 最大延迟（毫秒）:" + zookeeperserver.getmaxrequestlatency());
      system.out.println(role + " 每个客户端ip允许的最大连接数:" + zookeeperserver.getmaxclientcnxnsperhost());
      system.out.println(role + " 最大session超时（毫秒）:" + zookeeperserver.getmaxsessiontimeout());
      system.out.println(role + " 心跳时间（毫秒）:" + zookeeperserver.getticktime());
      system.out.println(role + " 版本:" + zookeeperserver.getversion());// 版本
      // 三个重置操作
//      zookeeperserver.resetlatency(); //重置min/avg/max latency statistics
//      zookeeperserver.resetmaxlatency(); //重置最大延迟统计
//      zookeeperserver.resetstatistics(); // 重置包和延迟所有统计
      system.out.println("==============所有客户端的连接信息===========");
      objectname connectionpattern = new objectname("org.apache.zookeeperservice:name0=replicatedserver_id?,name1=replica.?,name2=*,name3=connections,*");
      set<objectname> connectionsets = mbsc.querynames(connectionpattern, null);
      list<objectname> connectionlist = new arraylist<objectname>(connectionsets.size());
      connectionlist.addall(connectionsets);
      collections.sort(connectionlist);
      for (objectname connectionon : connectionlist) {
        system.out.println("=========================");
        connectionmxbean connectionbean = jmx.newmbeanproxy(mbsc, connectionon, connectionmxbean.class);
        system.out.println(" ip+port:" + connectionbean.getsourceip());//
        system.out.println(" sessionid:" + connectionbean.getsessionid());//
        system.out.println(" packetsreceived:" + connectionbean.getpacketsreceived());// 收到的包
        system.out.println(" packetssent:" + connectionbean.getpacketssent());// 发送的包
        system.out.println(" minlatency:" + connectionbean.getminlatency());//
        system.out.println(" avglatency:" + connectionbean.getavglatency());//
        system.out.println(" maxlatency:" + connectionbean.getmaxlatency());//
        system.out.println(" startedtime:" + connectionbean.getstartedtime());//
        system.out.println(" ephemeralnodes:" + connectionbean.getephemeralnodes().length);//
        system.out.println(" ephemeralnodes:" + arrays.aslist(connectionbean.getephemeralnodes()));//
        system.out.println(" outstandingrequests:" + connectionbean.getoutstandingrequests());//
        //connectionbean.resetcounters();
        //connectionbean.terminateconnection();
        //connectionbean.terminatesession();
      }
    }
    // close connection
    if (connector != null) {
      connector.close();
    }
  }
  /**
   * 建立连接
   * @param ip
   * @param jmxport
   * @return
   */
  public static mbeanserverconnection creatembeanserver(string ip,
      string jmxport, string username, string password) {
    try {
      string jmxurl = "service:jmx:rmi:///jndi/rmi://" + ip + ":"
          + jmxport + "/jmxrmi";
      // jmxurl
      jmxserviceurl serviceurl = new jmxserviceurl(jmxurl);
      map<string, string[]> map = new hashmap<string, string[]>();
      string[] credentials = new string[] { username, password };
      map.put("jmx.remote.credentials", credentials);
      connector = jmxconnectorfactory.connect(serviceurl, map);
      mbeanserverconnection mbsc = connector.getmbeanserverconnection();
      return mbsc;
    } catch (ioexception ioe) {
      ioe.printstacktrace();
      system.err.println(ip + ":" + jmxport + " 连接建立失败");
    }
    return null;
  }
  /**
   * 使用mbeanserver获取对象名为[objname]的mbean的[objattr]属性值
   * <p>
   * 静态代码: return mbeanserver.getattribute(objectname name, string attribute)
   * @param mbeanserver
   *      - mbeanserver实例
   * @param objname
   *      - mbean的对象名
   * @param objattr
   *      - mbean的某个属性名
   * @return 属性值
   */
  private static string getattribute(mbeanserverconnection mbeanserver,
      objectname objname, string objattr) {
    if (mbeanserver == null || objname == null || objattr == null)
      throw new illegalargumentexception();
    try {
      return string.valueof(mbeanserver.getattribute(objname, objattr));
    } catch (exception e) {
      return null;
    }
  }
}

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，谢谢大家对的支持。如果你想了解更多相关内容请查看下面相关链接