import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.ZooDefs.Ids;

import org.apache.zookeeper.ZooKeeper;

public class DistributedServer {

    //服务端启动时在zookeeper集群中 /services节点写入一个临时节点，该临时节点内容为该节点的名称。当服务端下线或者宕机时zookeeper会将该节点删除
　　 //一旦会话(Session)结束，临时节点将被自动删除

    private static final String connectString = "shizhan2:2183,shizhan3:2183,shizhan5:2183";

    private static final int sessionTimeout = 2000;

    private static final String parentNode = "/servers";

    private ZooKeeper zk = null;

    /**

     * 创建到zk的客户端连接

     * @throws Exception

     */

    public void getConnect() throws Exception {

        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

            @Override

            public void process(WatchedEvent event) {

                // 收到事件通知后的回调函数（应该是我们自己的事件处理逻辑）

                System.out.println(event.getType() + "---" + event.getPath());

                try {

                    zk.getChildren("/", true);

                } catch (Exception e) {

                }

            }

        });

    }

    /**

     * 向zk集群注册服务器信息

     * @param hostname

     * @throws Exception

     */

    public void registerServer(String hostname) throws Exception {

        String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        System.out.println(hostname + "is online.." + create);

    }

    /**

     * 业务功能

     * @throws InterruptedException

     */

    public void handleBussiness(String hostname) throws InterruptedException {

        System.out.println(hostname + "start working.....");

        Thread.sleep(Long.MAX_VALUE);

    }

    public static void main(String[] args) throws Exception {

        // 获取zk连接

        DistributedServer server = new DistributedServer();

        server.getConnect();

        // 利用zk连接注册服务器信息

        server.registerServer(args[0]);

        // 启动业务功能服务器

        server.handleBussiness(args[0]);

    }

}

package cn.itcast.bigdata.zkdist;

import java.util.ArrayList;

import java.util.List;

import org.apache.zookeeper.KeeperException;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.ZooKeeper;

public class DistributedClient {

    //客户端启动时会查找/services 下所有节点 并将节点名输出，当节点变化时，会重新查找所有节点，并将节点名输出

    private static final String connectString = "shizhan2:2183,shizhan3:2183,shizhan5:2183";

    private static final int sessionTimeout = 2000;

    private static final String parentNode = "/servers";

    // 注意:volatile是一个类型修饰符（type specifier），就像大家更熟悉的const一样，它是被设计用来修饰被不同线程访问和修改的变量

    private volatile List<String> serverList;

    private ZooKeeper zk = null;

    /**

     * 创建到zk的客户端连接

     * @throws Exception

     */

    public void getConnect() throws Exception {

        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

            @Override

            public void process(WatchedEvent event) {

                // 收到事件通知后的回调函数（应该是我们自己的事件处理逻辑）

                try {

                    //重新更新服务器列表，并且注册了监听

                    System.out.println("重新更新服务器列表，并且注册了监听");

                    getServerList();

                } catch (Exception e) {

                }

            }

        });

    }

    /**

     * 获取服务器信息列表

     * @throws Exception

     */

    public void getServerList() throws Exception {

        // 获取服务器子节点信息，并且对父节点进行监听

        List<String> children = zk.getChildren(parentNode, true);

        // 先创建一个局部的list来存服务器信息

        List<String> servers = new ArrayList<String>();

        for (String child : children) {

            // child只是子节点的节点名

            byte[] data = zk.getData(parentNode + "/" + child, false, null);

            servers.add(new String(data));

        }

        // 把servers赋值给成员变量serverList，提供给各业务线程使用

        serverList = servers;

        //打印服务器列表

        System.out.println("=========打印服务器列表=========");

        System.out.println(serverList);

    }

    /**

     * 业务功能

     *

     * @throws InterruptedException

     */

    public void handleBussiness() throws InterruptedException {

        System.out.println("client start working.....");

        Thread.sleep(Long.MAX_VALUE);

    }

    public static void main(String[] args) throws Exception {

        // 获取zk连接

        DistributedClient client = new DistributedClient();

        client.getConnect();

        // 获取servers的子节点信息（并监听），从中获取服务器信息列表

        client.getServerList();

        // 业务线程启动

        client.handleBussiness();

    }

}