zookeeper实现分布式应用系统服务器上下线动态感知程序、监听机制与守护线程

/**
 * 服务端程序
 * @author
 *
 */
public class DistributedServer {
    private static final String connectionString = "192.168.47.141:2181";
    public static final Integer sessionTimeout = 2000;
    public static ZooKeeper zkClient = null;
    /**
     * 获取zookeeper连接
     * @throws Exception
     */
    public void getConnection() throws Exception{
        zkClient = new ZooKeeper(connectionString, sessionTimeout, new Watcher(){
            //收到事件通知后的回调函数（应该是我们自己的事件处理逻辑）
            public void process(WatchedEvent event) {
                System.out.println(event.getType()+","+event.getPath());
                try {
                    //为了能一直监听,调用一次注册一次
                    zkClient.getChildren("/", true);
                }catch(Exception e){
                }
            }});
    }
    /**
     * 注册服务器信息
     * @param hostname 注册的服务器名
     * @throws Exception
     */
    public void registerServer(String hostname) throws Exception{
        //创建的是带序号的临时节点 生成的节点像/servers/server000001,/servers/server000002等
        //节点数据即为注册的主机名
        String path = zkClient.create("/server", hostname.getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(hostname+" --上线了-- "+path);
    }
    /**
     * 服务器注册完后，执行业务逻辑
     * @param hostname
     * @throws IOException
     */
    public void executeBusiness(String hostname) throws IOException {
        System.out.println(hostname+"开始工作了!");
        System.in.read();
    }
    public static void main(String[] args) throws Exception {
        //获取zookeeper连接
        DistributedServer server = new DistributedServer();
        server.getConnection();
        //服务器上线，完成注册
        Scanner scanner = new Scanner(System.in);
        System.out.println("输入hostname");
        String hostname = scanner.nextLine();
        server.registerServer(hostname);
        //执行业务逻辑
        server.executeBusiness(hostname);
    }
}

/*
 * 客户端程序
 */
public class DistributeClient {
    private static final String connectionString = "192.168.47.141:2181";
    public static final Integer sessionTimeout = 2000;
    public static ZooKeeper zkClient = null;
    public static final String parentNode = "/";
    //注意：加volatile的意义何在?使得多线程看到的服务器列表一致而不会拷贝到自己的工作空间
    public volatile List<String> serverList = new ArrayList<String>();
    /**
     * 获取zookeeper连接
     * @throws Exception
     */
    public void getConnection() throws Exception{
        zkClient = new ZooKeeper(connectionString, sessionTimeout, new Watcher(){
            //收到事件通知后的回调函数（应该是我们自己的事件处理逻辑）
            public void process(WatchedEvent event) {
                System.out.println(event.getType()+","+event.getPath());
                try {
                    //重新获取(更新)服务器列表,并进行监听
                    getServerList();
                }catch(Exception e){
                }
            }});
    }
    /**
     * 获取服务器列表信息，并对父节点进行监听
     * @throws Exception
     */
    public void getServerList() throws Exception{
        //获取服务器列表，并对父节点进行监听
        //getChildren()相对于命令行 ls /znode,对子节点进行监听
        List<String> children = zkClient.getChildren(parentNode, true);
        //创建临时集合，将子节点存入
        List<String> childrenList = new ArrayList<String>();
        for (String child : children) {
            byte[] data = zkClient.getData(parentNode+child, false, null);
            childrenList.add(new String(data));
        }
        //将临时集合中的节点赋给服务器列表serverList,以便业务线程使用
        serverList = childrenList;
        System.out.println(serverList);
    }
    /**
     * 业务功能
     * @throws Exception
     */
    public void executeBusiness() throws Exception{
        System.out.println("获取的服务器列表:"+serverList);
        System.out.println("客户端开始工作了...");
        System.in.read();
    }
    public static void main(String[] args) throws Exception {
        //获取zookeeper连接
        DistributeClient client = new DistributeClient();
        client.getConnection();
        //获取服务器列表
        client.getServerList();
        //业务功能
        client.executeBusiness();
    }
}