Zookeeper实现分布式集群监控

Zookeeepr实现分布式集群监控

Zookeeper中节点有两种：临时节点和永久节点

从类型上看节点又可以分为四种节点类型：PERSIST，PERSIST_SEQUENTIAL,EPHEMERAL,EPHEMERAL_SEQUENTIAL

临时节点有一个特点：当创建临时节点的程序停掉之后，这个临时节点就会消失。

监视器的特点：可以给zk中的节点注册监视器，见识这个节点的变化情况。

监视器注册一次，只能使用一次，多次使用就要多次注册。

我们利用这个Zookeeper的临时节点特性+监视器(Watch)来实现分布式集群监控

我们在/monitor（永久节点）下创建临时节点。

实际上，zookeeper的sdk不是特别好用，很多边界情况需要用户自己处理。curator是对Zookeeper的sdk进行了封装，所以说使用curator操作Zookeeper更加方便

在maven官网找到curator的依赖

<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-framework,支持zookeeper3.4.6-->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.10.0</version>
</dependency>

我们通过curator来使用zookeeper，那么我们就必须知道如何使用curator来连上zookeeper，下面代码是官方文档所给出

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3)CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
client.start();

开始我们的代码

TestCurator.java ,实现功能：创建Zookeeper临时节点

package zkdemo;

import java.net.InetAddress;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;  

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Test;


public class TestCurator {
    
    @Test
    public void test1() throws Exception{
        //1000：表示curator链接zk的时候超时时间是多少 3：表示链接zk的最大重试次数
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        String connectString = "djt1:2181,djt2:2181,djt3:2181,djt4:2181,djt5:2181";
        int sessionTimeoutMs = 5000;//这个值只能在4000-40000ms之间表示链接断开后多长时间临时节点会小时
        int connectionTimeoutMs = 3000;//获取链接的超时时间
        //创建一个zk连接
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, sessionTimeoutMs
                ,connectionTimeoutMs,retryPolicy);
        
        client.start();
        
        InetAddress localHost = InetAddress.getLocalHost();
        String ip = localHost.getHostAddress();
        
        client.create().creatingParentsIfNeeded()
        .withMode(CreateMode.EPHEMERAL)//指定节点类型
        .withACL(Ids.OPEN_ACL_UNSAFE)//指定设置节点权限信息
        .forPath("/monitor/"+ip);//指定节点名称
        
        while(true)
        {
            ;
        }
    }
}

ZkNodeWatcher.java 实现功能:注册watcher，监视节点的变化

package zk;

import java.util.List;
import java.util.ArrayList;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;



public class ZkNodeWatcher implements Watcher{
    CuratorFramework client;
    List<String> childrenList = new ArrayList<String>();
    List<String> newChildrenList = new ArrayList<String>();

    public  ZkNodeWatcher(){
        //1000：表示curator链接zk的时候超时时间是多少 3：表示链接zk的最大重试次数
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        String connectString = "djt1:2181,djt2:2181,djt3:2181,djt4:2181,djt5:2181";
        int sessionTimeoutMs = 5000;//这个值只能在4000-40000ms之间表示链接断开后多长时间临时节点会消失
        int connectionTimeoutMs = 3000;//获取链接的超时时间
        //创建一个zk连接
        client = CuratorFrameworkFactory.newClient(connectString, sessionTimeoutMs
                ,connectionTimeoutMs,retryPolicy);
        client.start();
        
        //监视monitor节点,获取下面的所有子节点的变化情况
        try {
            childrenList = client.getChildren().usingWatcher(this).forPath("/monitor");
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
        
    

    /**
     * 实现一个zk监视器，监视某个节点的变化情况
     * 
     * 这个监视程序需要一直运行
     * @CPH
     */
    
    public void process(WatchedEvent event) {
        System.out.println("我被调用了");
        try {
            newChildrenList = client.getChildren().usingWatcher(this).forPath("/monitor");
            for(String ip : childrenList)
            {
                if(!newChildrenList.contains(ip)){
                    System.out.println("节点消失了"+ip);
                    //TODO 给管理员发送短信什么的
                    
                }
            }
            
        for(String ip : newChildrenList){
            if(!childrenList.contains(ip)){
                System.out.println("节点新增"+ip);
            }
        }
        //重要
        childrenList = newChildrenList;
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        
        
    }
    
    public void start(){
        while (true){;}
        }

    public static void main(String[] args) {
        ZkNodeWatcher watcher = new ZkNodeWatcher();
        watcher.start();
    }
}

我们先开启Zookeeper集群，启动/bin/zkCli.sh,然后启动这2个集群，我们可以看到由对应的ip目录，这个ip不是虚拟机的ip，而是本地的ip，同时我们console下看到

然后暂停TestCurator.java,不一会儿，就会看到

这样，通过Zookeeper实现分布式集群监控的功能就完成了！