zookeeper大规模分布式集群中任一单点设备上线下线心跳感知系统（一）

分布式集群中，为了能全局感知任一单点设备的存活状态，经常有心跳感知系统的设计需求，要实现这样的心跳感知。
常规的做法无法就是保持一个Socket长连接或者http短连接，但是这样的实现手法往往扩展性极差，且问题非常多，维护成本很高。
而zookeeper恰恰就是这种分布式集群大规模设备心跳感知系统的最佳“框架”性解决方案。现在假设一个客户端，它需要实时监测分布式集群中的设备节点上线/下线/掉线情况，那么它可以（假设叫做 程序A）：

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class Main {
    private ZooKeeper zk;
    private String DEVICE_PATH = "/devices";

    public static void main(String[] args) {
        Main main = new Main();
        try {
            main.test();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void test() throws Exception {
        int t = 10;
        zk = new ZooKeeper(getAddress(), t * 1000, new MyWatcher());

        System.out.println("开始连接...");
        while (!zk.getState().isConnected()) {
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println("连接建立");

        //创建/devices节点路径的工作可以手动。不必一定代码完成。
        if (!checkNodeExist(DEVICE_PATH)) {
            String path = zk.create(DEVICE_PATH, "server".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("创建:" + path);
        }

        synchronized (this) {
            wait();
        }
    }

    private class MyWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            if (!checkNodeExist(DEVICE_PATH)) {
                return;
            }

            System.out.println("---------");

            try {
                List<String> children = zk.getChildren(DEVICE_PATH, true);

                for (String c : children) {
                    String path = DEVICE_PATH + "/" + c;
                    byte[] data = zk.getData(path, false, null);
                    System.out.println("设备在线:" + path + " @ " + new String(data));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private String getAddress() {
        String ip = "127.0.0.1";
        return ip + ":2181," + ip + ":2182," + ip + ":2183";
    }

    private boolean checkNodeExist(String path) {
        boolean b = false;

        try {
            Stat stat = zk.exists(path, true);
            b = stat == null ? false : true;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return b;
    }
}

注意，上面的程序A代码不要把它理解为服务器端，事实上程序A可以运行在分布式集群中的任何设备终端上，它实现的一个功能是捕获和感知任何设备的上线和下线情况。当有新设备接入程序A所在的集群中时候，就被MyWatcher捕获和感知到，同样，只要有设备断开、离开集群，也会被MyWatcher捕获和感知到。

 

 

下面再写一个程序B，程序B模拟一个新设备接入分布式集群：

import org.apache.zookeeper.*;

import java.util.concurrent.TimeUnit;

public class Main {
    private ZooKeeper zooKeeper;
    private String DEVICE_PATH = "/devices";

    public static void main(String[] args) {
        //初始化log4j，zookeeper否则报错。
        //org.apache.log4j.BasicConfigurator.configure();

        try {
            Main app = new Main();
            app.zk();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void zk() throws Exception {
        zooKeeper = new ZooKeeper(getAddress(), 10 * 1000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                //System.out.println(event.toString());
            }
        });

        System.out.println("开始连接...");
        while (!zooKeeper.getState().isConnected()) {
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println("连接建立");

        login();

        synchronized (this) {
            wait();
        }
    }

    private void login() throws Exception {
        //创建节点。
        String path = zooKeeper.create(DEVICE_PATH + "/client", "ip:127.0.0.1".getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        byte[] data = zooKeeper.getData(path, null, null);
        System.out.println(path + " => " + new String(data));
        System.out.println("注册设备成功");
    }

    private String getAddress() {
        String ip = "127.0.0.1";
        return ip + ":2181," + ip + ":2182," + ip + ":2183";
    }
}

运行程序B，相当于分布式集群中有新设备进入，那么程序B代表的设备就会被程序A捕获。若程序B离开、断开与集群的连接，那么同样可以被程序A感知。

实际上，把程序A的心跳、感知功能代码(其实就是程序A的MyWatcher)写入到程序B中，那么程序A，和程序B就可以互相感知，这样就做到在分布式环境中，设备心跳，上线/下线的互相感知系统。