个人随笔
目录
四、zookeeper使用场景之：分布式集群管理
2020-12-03 16:29:19

一、背景

我们再分布式集群环境中，可能会有多个节点，作为管理员，有必要知道线上有多少个节点在运行中，以及节点变化情况和每个节点的内存、CPU、磁盘的使用情况，若是有超出预警值需要及时通知我们，发短信或者邮件，这样子就可以快速响应生产故障或者预防生产故障。

为了实现上面的目标，这里打算借助zookeeper来监控每个节点的存活情况已经内存CPU的使用情况，关键点是用zookeeper的临时序号节点和监听。下面举个例子吧。

二、例子

启动几个节点，节点启动的时候就上报道zookeeper，然后定时上报自己内存CPU等资源的使用情况。监控程序监控节点的变化情况以及节点内容的变化情况，有异常则发邮件给管理员。

三、代码实例

这里不用zookeeper的原生API，原生API实现起来比较复杂。比如要自己去实现循环监听的情况。所以这里用zkClient来。

1、新建maven项目，pom.xml加入如下依赖

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>com.google.code.gson</groupId>
  4.         <artifactId>gson</artifactId>
  5.         <version>2.8.5</version>
  6.     </dependency>
  7.     <!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
  8.     <dependency>
  9.         <groupId>com.101tec</groupId>
  10.         <artifactId>zkclient</artifactId>
  11.         <version>0.11</version>
  12.     </dependency>
  13. </dependencies>

2、建立节点类Node

  1. public class Node {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         new ZkClientMsg().init("2");
  4.         System.out.println("执行正常业务逻辑...");
  5.         Thread.sleep(Long.MAX_VALUE);
  6.     }
  7. }

每个节点在初始化的似乎都调用ZkClientMsg类来将信息上报道zookeeper，然后再执行自己的业务逻辑。

3、ZkClientMsg

该类就是实现了链接zookeeper以及创建临时序号节点，用临时序号节点的特性：当节点挂掉后临时序号节点也会自动清除来监控节点的存活情况，以及定时上报自己的服务器相关信息，代码如下：

  1. package com.suibibk.zookeeper;
  3. import java.io.IOException;
  4. import java.text.SimpleDateFormat;
  5. import java.util.Date;
  6. import java.util.HashMap;
  7. import java.util.List;
  8. import java.util.Map;
  10. import org.I0Itec.zkclient.ZkClient;
  11. import org.apache.zookeeper.CreateMode;
  13. import com.google.gson.Gson;
  15. public class ZkClientMsg {
  16.     private   ZkClient zkClient;
  17.     //根节点：为持久及诶单
  18.     private String rootPath = "/service";
  19.     //子节点：为临时序号节点
  20.     private String servicePath = "/service/service";
  21.     private String NODE = "1";
  22.     //获取当前节点的名称
  23.     private String nowPath = "";
  24.     public  void init(String node){
  25.         try {
  26.             //初始化
  27.             System.out.println("Node start..."+node);
  28.             NODE = node;
  29.             //上报Zookeeper
  30.             initZookeeper();
  31.             //创建根节点:根节点用的是持久节点
  32.             createRoot();
  33.             //创建当前节点：为临时序号节点
  34.             createCurrentNode();
  36.             //定时上报IP，内存，cpu，磁盘等使用情况
  37.             new Thread(new Runnable() {
  39.                 public void run() {
  40.                     while(true) {
  41.                         try {
  42.                             updateCurrentNode();
  43.                             //5秒钟上报一次
  44.                             Thread.sleep(5*1000l);
  45.                         } catch (InterruptedException e) {
  46.                             // TODO Auto-generated catch block
  47.                             e.printStackTrace();
  48.                         }
  49.                     }
  50.                 }
  51.             }).start();
  52.             System.out.println("Node finish");
  53.         } catch (Exception e) {
  54.             // TODO Auto-generated catch block
  55.             e.printStackTrace();
  56.         }
  57.     }
  59.     private void createRoot() {
  60.         //检查根节点是否存在
  61.         try {
  62.             Boolean result = zkClient.exists(rootPath);
  64.             if(!result) {
  65.                 //根节点不存在，创建
  66.                 System.out.println("根节点不存在，创建持久节点");
  67.                 create(rootPath, "service", CreateMode.PERSISTENT);
  68.             }
  69.         } catch (Exception e) {
  70.             e.printStackTrace();
  71.         }
  73.     }
  74.     private String create(String path,String data,CreateMode mode) {
  75.         try {
  76.             String path2 = zkClient.create(path,data,mode);
  77.             return path2;
  78.         } catch (Exception e) {
  79.             e.printStackTrace();
  80.         }
  81.         return "";
  82.     }
  83.     public void initZookeeper() throws IOException {
  84.         String connectString = "192.168.209.4:2181";
  85.         zkClient = new ZkClient(connectString, 50*1000);
  86.     } 
  88.     /**
  89.      * 创建临时节点
  90.      */
  91.     private void createCurrentNode() {
  92.         //获取用户的IP，CPU,RAN,DISK
  93.         String result = getOSInfo();
  94.         //创建临时序号节点
  95.         String path = create(servicePath, result, CreateMode.EPHEMERAL_SEQUENTIAL);
  96.         nowPath=path;
  97.     }
  98.     /**
  99.      *更新当前临时节点
  100.      */
  101.     private void updateCurrentNode() {
  102.         //获取用户的IP，CPU,RAN,DISK
  103.         String result = getOSInfo();
  104.         //创建临时序号节点,这里要返回节点名称
  105.         //-1表示不考虑当前版本信息
  106.         try {
  107.             zkClient.writeData(nowPath, result);
  108.         } catch (Exception e) {
  109.             e.printStackTrace();
  110.         }
  112.     }
  113.     /**
  114.      * 获取系统信息
  115.      * @return
  116.      */
  117.     private String getOSInfo() {
  118.         String str0 = GetSystemInfo.getLocalIp();
  119.         String str1 = GetSystemInfo.getMemery();
  120.         List<String> strs = GetSystemInfo.getDisk();
  121.         String str3 = GetSystemInfo.getCpuRatioForWindows();
  122.         Map<String,Object> map = new HashMap<String,Object>();
  123.         map.put("TIME",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
  124.         map.put("IP",str0);
  125.         map.put("CPU",str3);
  126.         map.put("RAN",str1);
  127.         map.put("DISK",strs);
  128.         map.put("NODE",NODE);
  129.         Gson gson = new Gson();
  130.         String result = gson.toJson(map);
  131.         return result;
  133.     }
  135. }

4、GetSystemInfo

这是一个获取服务器相关信息的工具类，CPU只支持在win上的获取,这里只是举一个例子，所以就不要求这么精确了，如果是在正式环境使用，建议用专门的工具类来：

  1. package com.suibibk.zookeeper;
  3. import java.io.File;
  4. import java.io.InputStreamReader;
  5. import java.io.LineNumberReader;
  6. import java.lang.management.ManagementFactory;
  7. import java.net.InetAddress;
  8. import java.net.UnknownHostException;
  9. import java.util.ArrayList;
  10. import java.util.List;
  12. import com.sun.management.OperatingSystemMXBean;
  14. public class GetSystemInfo {
  15.     private static final int CPUTIME = 500;  
  17.     private static final int PERCENT = 100;  
  19.     private static final int FAULTLENGTH = 10;  
  21.     /**
  22.      * 获取内存使用情况
  23.      * @return
  24.      */
  25.     public static String getMemery() {
  26.          OperatingSystemMXBean osmxb = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
  27.          //物理内存
  28.          long totalPhysicalMemory = osmxb.getTotalPhysicalMemorySize();
  29.          //可用物理内存  
  30.          long freePhysicalMemorySize = osmxb.getFreePhysicalMemorySize(); 
  31.          //虚拟内存  
  32.          long totalvirtualMemory = osmxb.getTotalSwapSpaceSize();
  33.          //可用虚拟内存
  34.          long freeSwapSpaceSize = osmxb.getFreeSwapSpaceSize();
  36.         // System.out.println("物理内存：" + totalPhysicalMemory / 1024 / 1024 + "MB");
  37.         // System.out.println("可用物理内存：" + freePhysicalMemorySize / 1024 / 1024 + "MB");
  38.         // System.out.println("虚拟内存 ：" + totalvirtualMemory / 1024 / 1024 + "MB");
  39.         // System.out.println("可用虚拟内存：" + freeSwapSpaceSize / 1024 / 1024 + "MB");
  40.          Double compare = Double.valueOf((1 - freePhysicalMemorySize * 1.0 / totalPhysicalMemory) * 100);  
  41.          String str = "物理内存已使用:" +compare.intValue()+ "%";  
  42.          return str;
  43.     }
  45.     /**
  46.      * 获取文件系统使用率  
  47.      * @return
  48.      */
  49.     public static List<String> getDisk()  
  50.     {  
  51.         // 操作系统  
  52.         List<String> list = new ArrayList<String>();  
  53.         for (char c = 'A'; c <= 'Z'; c++)  
  54.         {  
  55.             String dirName = c + ":/";  
  56.             File win = new File(dirName);  
  57.             if (win.exists())  
  58.             {  
  59.                 long total = (long)win.getTotalSpace();  
  60.                 long free = (long)win.getFreeSpace();  
  61.                 Double compare = Double.valueOf((1 - free * 1.0 / total) * 100);  
  62.                 String str = c + ":盘  已使用 " + compare.intValue() + "%";  
  63.                 list.add(str);  
  64.             }  
  65.         }  
  66.         return list;  
  67.     }
  68.     /**
  69.      * 获取Windows下CPU的使用率  
  70.      * @return
  71.      */
  72.     public static String getCpuRatioForWindows()  
  73.     {  
  74.         try  
  75.         {  
  76.             String procCmd =  
  77.                 System.getenv("windir")  
  78.                     + "\\system32\\wbem\\wmic.exe process get Caption,CommandLine,KernelModeTime,ReadOperationCount,ThreadCount,UserModeTime,WriteOperationCount";  
  79.             // 取进程信息 
  80.             long[] c0 = readCpu(Runtime.getRuntime().exec(procCmd));  
  81.             Thread.sleep(CPUTIME);  
  82.             long[] c1 = readCpu(Runtime.getRuntime().exec(procCmd));  
  83.             if (c0 != null && c1 != null)  
  84.             {  
  85.                 long idletime = c1[0] - c0[0];  
  86.                 long busytime = c1[1] - c0[1];  
  87.                 return "CPU使用率:" + Double.valueOf(PERCENT * (busytime) * 1.0 / (busytime + idletime)).intValue() + "%";  
  88.             }  
  89.             else  
  90.             {  
  91.                 return "CPU使用率:" + 0 + "%";  
  92.             }  
  93.         }  
  94.         catch (Exception ex)  
  95.         {  
  96.             ex.printStackTrace();  
  97.             return "CPU使用率:" + 0 + "%";  
  98.         }  
  99.     }  
  100.     /**
  101.      * 获取IP
  102.      * @return
  103.      */
  104.     public static String getLocalIp() {
  105.         InetAddress addr = null;
  106.         try {
  107.             addr = InetAddress.getLocalHost();
  108.         } catch (UnknownHostException e) {
  109.             throw new RuntimeException(e);
  110.         }
  111.         return addr.getHostAddress();
  112.     }
  113.     //读取cpu相关信息  
  114.     private static long[] readCpu(final Process proc)  
  115.     {  
  116.         long[] retn = new long[2];  
  117.         try  
  118.         {  
  119.             proc.getOutputStream().close();  
  120.             InputStreamReader ir = new InputStreamReader(proc.getInputStream());  
  121.             LineNumberReader input = new LineNumberReader(ir);  
  122.             String line = input.readLine();  
  123.             if (line == null || line.length() < FAULTLENGTH)  
  124.             {  
  125.                 return null;  
  126.             }  
  127.             int capidx = line.indexOf("Caption");  
  128.             int cmdidx = line.indexOf("CommandLine");  
  129.             int rocidx = line.indexOf("ReadOperationCount");  
  130.             int umtidx = line.indexOf("UserModeTime");  
  131.             int kmtidx = line.indexOf("KernelModeTime");  
  132.             int wocidx = line.indexOf("WriteOperationCount");  
  133.             long idletime = 0;  
  134.             long kneltime = 0;  
  135.             long usertime = 0;  
  136.             while ((line = input.readLine()) != null)  
  137.             {  
  138.                 if (line.length() < wocidx)  
  139.                 {  
  140.                     continue;  
  141.                 }  
  142.                 // 字段出现顺序：Caption,CommandLine,KernelModeTime,ReadOperationCount,  
  143.                 // ThreadCount,UserModeTime,WriteOperation  
  144.                 String caption = substring(line, capidx, cmdidx - 1).trim();  
  145.                 String cmd = substring(line, cmdidx, kmtidx - 1).trim();  
  146.                 if (cmd.indexOf("wmic.exe") >= 0)  
  147.                 {  
  148.                     continue;  
  149.                 }  
  150.                 String s1 = substring(line, kmtidx, rocidx - 1).trim();  
  151.                 String s2 = substring(line, umtidx, wocidx - 1).trim();  
  152.                 if (caption.equals("System Idle Process") || caption.equals("System"))  
  153.                 {  
  154.                     if (s1.length() > 0)  
  155.                         idletime += Long.valueOf(s1).longValue();  
  156.                     if (s2.length() > 0)  
  157.                         idletime += Long.valueOf(s2).longValue();  
  158.                     continue;  
  159.                 }  
  160.                 if (s1.length() > 0)  
  161.                     kneltime += Long.valueOf(s1).longValue();  
  162.                 if (s2.length() > 0)  
  163.                     usertime += Long.valueOf(s2).longValue();  
  164.             }  
  165.             retn[0] = idletime;  
  166.             retn[1] = kneltime + usertime;  
  167.             return retn;  
  168.         }  
  169.         catch (Exception ex)  
  170.         {  
  171.             ex.printStackTrace();  
  172.         }  
  173.         finally  
  174.         {  
  175.             try  
  176.             {  
  177.                 proc.getInputStream().close();  
  178.             }  
  179.             catch (Exception e)  
  180.             {  
  181.                 e.printStackTrace();  
  182.             }  
  183.         }  
  184.         return null;  
  185.     }  
  187.     /** 
  188.     * 由于String.subString对汉字处理存在问题（把一个汉字视为一个字节)，因此在 包含汉字的字符串时存在隐患，现调整如下： 
  189.     * @param src 要截取的字符串 
  190.     * @param start_idx 开始坐标（包括该坐标) 
  191.     * @param end_idx 截止坐标（包括该坐标） 
  192.     * @return 
  193.     */  
  194.     private static String substring(String src, int start_idx, int end_idx)  
  195.     {  
  196.         byte[] b = src.getBytes();  
  197.         String tgt = "";  
  198.         for (int i = start_idx; i <= end_idx; i++)  
  199.         {  
  200.             tgt += (char)b[i];  
  201.         }  
  202.         return tgt;  
  203.     }
  204.     public static void main(String[] args) {
  205.         String str0 = getLocalIp();
  206.         System.out.println(str0);
  207.         String str1 = getMemery();
  208.         System.out.println(str1);
  209.         List<String> strs = getDisk();
  210.         for (String str : strs) {
  211.             System.out.println(str);
  212.         }
  213.         String str3 = getCpuRatioForWindows();
  214.         System.out.println(str3);
  215.     }
  216. }

上面节点就基本上完成啦，只需要在Node上启动就可以了。

5、监控器：ZkClientMonitor

下面这里列一下监控器，监控器的实现也很简单，就是监控节点下面的子节点数目的改变，子节点就是每个节点的临时序号节点。然后再监听每个子节点的数据变化，如果超过规定值就预警。

  1. package com.suibibk.zookeeper;
  2. import java.util.ArrayList;
  3. import java.util.List;
  4. import java.util.Map;
  6. import org.I0Itec.zkclient.IZkChildListener;
  7. import org.I0Itec.zkclient.IZkDataListener;
  8. import org.I0Itec.zkclient.ZkClient;
  10. import com.google.gson.Gson;
  11. /**
  12.  * @author 小林同学
  13.  */
  14. public class ZkClientMonitor {
  15.      final static  String path ="/service";
  16.      private static ZkClient zkClient;
  17.      public static void main(String[] args) throws InterruptedException {
  18.          String connectString = "192.168.209.4:2181";
  19.          zkClient = new ZkClient(connectString, 50*1000);
  20.          initWatch(path);
  21.          Thread.sleep(Long.MAX_VALUE);
  22.     }
  23.      /**
  24.       * 初始化监听器
  25.       * @param nodePath
  26.       */
  27.      public static void initWatch(String path) {
  28.          //1、先移除所有监听
  29.          zkClient.unsubscribeAll();
  30.          //2、添加子节点变更监听
  31.          List<String> paths = zkClient.subscribeChildChanges(path, new IZkChildListener() {
  32.              public void handleChildChange(String arg0, List<String> arg1) throws Exception {
  33.                  // TODO Auto-generated method stub
  34.                  //System.out.println("重新初始化监听器："+arg0);
  35.                  List<String> strs = new ArrayList<String>();
  36.                  for (String string : arg1) {
  37.                      String str = zkClient.readData(arg0+"/"+string);
  38.                      strs.add(str);
  39.                 }
  40.                  System.out.println("节点数目有变化，当前存活节点为："+strs+"请留意");
  41.                  initWatch(arg0);
  42.              }
  43.          });
  44.          for (String string : paths) {
  45.              //System.out.println("监听path："+string+"的数据变化");
  46.              //这里进行数据变化监听
  47.              zkClient.subscribeDataChanges(path+"/"+string, new IZkDataListener() {
  48.                 public void handleDataDeleted(String arg0) throws Exception {
  49.                 }
  50.                 public void handleDataChange(String arg0, Object arg1) throws Exception {
  51.                     //System.out.println("arg0:"+arg0+"change data");
  52.                     System.out.println("节点数据变化:"+arg1);
  53.                     //这里对数据进行监控判断，若有数据变化超过预警值则发邮件通知管理员
  54.                     Map map = new Gson().fromJson(arg1.toString(), Map.class);
  55.                     //获取内存
  56.                     String ip = (String) map.get("IP");
  57.                     String ran = (String) map.get("RAN");
  58.                     //System.out.println("ran："+ran);
  59.                     int r = Integer.parseInt(ran.replace("%", "").replaceAll("物理内存已使用:", ""));
  60.                     //System.out.println("r："+r);
  61.                     if(r>60) {
  62.                         System.out.println("IP:"+ip+"的内存使用高达"+r+"%请留意");
  63.                     }
  64.                 }
  65.             });
  66.          }
  69.      }
  70. }

要重新移除监听的原因是因为我这没有具体的对哪些节点是新增的做筛选，实际实现中可以先保存为一个变量，然后在节点数目变化的情况下去匹对，只对新增的节点进行监控数据的改变，这样效果更好，不过我这里只是举个例子而已，为了防止已经加了监听的节点重新加监听我这里就直接先移除所有监听，为什么不移除单个监听的呢？原因如下：

注：我这里采取的是如果有变化就移除该监控节点的所监听然后再初始化监听，因为我尝试过用zkClient.unsubscribeDataChanges(arg0, arg1)这种只移除对应监听的方法完全不生效，就算用zookeeper原生API的removeAllWatchs也完全没有效果，目前还没有找到解决方案，希望有知道的可以告知我一下。

四、测试

测试的话就比较简单，这里先启动节点，然后启动监控，会发现打印如下日志：

  1. 节点数据变化:{"NODE":"1","IP":"192.168.209.1","CPU":"CPU使用率:7%","TIME":"2020-12-03 16:11:46","DISK":["C:盘  已使用 45%","D:盘  已使用 29%","E:盘  已使用 50%"],"RAN":"物理内存已使用:70%"}
  2. IP:192.168.209.1的内存使用高达70%请留意

然后我再启动一个节点。日志会打印如下

  1. 节点数目有变化，当前存活节点为：[{"NODE":"2","IP":"192.168.209.1","CPU":"CPU使用率:5%","TIME":"2020-12-03 16:12:14","DISK":["C:盘  已使用 45%","D:盘  已使用 29%","E:盘  已使用 50%"],"RAN":"物理内存已使用:71%"}, {"NODE":"1","IP":"192.168.209.1","CPU":"CPU使用率:10%","TIME":"2020-12-03 16:12:09","DISK":["C:盘  已使用 45%","D:盘  已使用 29%","E:盘  已使用 50%"],"RAN":"物理内存已使用:71%"}]请留意
  2. 节点数据变化:{"NODE":"2","IP":"192.168.209.1","CPU":"CPU使用率:5%","TIME":"2020-12-03 16:12:17","DISK":["C:盘  已使用 45%","D:盘  已使用 29%","E:盘  已使用 50%"],"RAN":"物理内存已使用:71%"}
  3. IP:192.168.209.1的内存使用高达71%请留意

表明监控到了数目的变化以及对新的节点数据的变化进行监听，实现了目标，收工。

 263

啊！这个可能是世界上最丑的留言输入框功能~


当然，也是最丑的留言列表

有疑问发邮件到 : suibibk@qq.com 侵权立删
Copyright : 个人随笔   备案号 : 粤ICP备18099399号-2