博客

#以创建Maven项目为例:

#POM文件导入依赖或者去仓库搜索相关所需版本:

本人常用仓库地址:

http://mvnrepository.com/

<dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.5.2-alpha</version>
    </dependency>

#如下为拉下相关jar依赖包

####服务端####

public class DistributedServer {
    //三台服务器主机名+端口分别用","隔开;
    private static final String connectString = "hadoop:2181,hadoop01:2181,hadoop02:2181";
    //超时设置
    private static final int sessionTimeout = 2000;
    //父节点
    private static final String parentNode = "/servers";

    private ZooKeeper zk = null;

    /**
     * 获取zk连接
     * @throws IOException
     */
    public void getConnect() throws IOException{
        /**
         * Watcher 监听器
         * 每监听一次之后都需要重新注册
         */
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                try {
                    //重新监听
                    zk.getChildren("/", true);
                    System.out.println("[server : " + event.getType() + ":" + event.getPath() + "]");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 向ZooKeeper集群注册服务器信息
     * @throws InterruptedException 
     * @throws KeeperException 
     */
    @SuppressWarnings("unused")
    private void registerServer(String hostName) throws KeeperException, InterruptedException{
        /**
         *  param1 : 要创建的节点路径 
         *  param2 : 节点的数据,必须是byte类型.
         *  param3 : 节点的权限(枚举类型) 
         *  param4 : 创建的类别(枚举类型)
         * 
         */
        String create = zk.create(parentNode + "/server", hostName.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL );
        System.out.println(hostName + " register ZooKeeper , Create Node Info :" + create);
    }

    private void handleBussiness(String hostName) throws InterruptedException{
        System.out.println(hostName + " start working.....");
        Thread.sleep(Long.MAX_VALUE);
    }

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        DistributedServer server = new DistributedServer();
        //获取ZK连接
        server.getConnect();
        //利用ZK连接注册服务器信息
        server.registerServer(args[0]);
        //业务功能
        server.handleBussiness(args[0]);
    }

}

####客户端####

public class DistributedClient {

    private static final String connectString = "hadoop:2181,hadoop01:2181,hadoop02:2181";
    private static final int sessionTimeout = 2000;
    private static final String parentNode = "/servers";
    /**
     *  注意:加volatile的意义何在？
     */
    private volatile List<String> serverList;

    //ZooKeeper;
    private ZooKeeper zk = null;

    /**
     * 获取zk客户端连接
     * @throws IOException
     */
    public void getConnect() throws IOException{
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                try {
                    //重新更新服务器列表，并且注册了监听
                    getServerList();
                    System.out.println("[client : " + event.getType() + ":" + event.getPath() + "]");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 获取服务器信息列表
     * @throws InterruptedException 
     * @throws KeeperException 
     */
    public void getServerList() throws KeeperException, InterruptedException{
        //获取服务器节点信息,并监听
        List<String> childrens = zk.getChildren(parentNode, true);
        //获取子节点下服务器列表
        List<String> servers = new ArrayList<String>();
        for (String serverNode : childrens) {
            //获取服务器节点下服务器名称
            byte[] data = zk.getData(parentNode + "/" + serverNode, false, null);
            servers.add(new String(data));
        }
        serverList = servers;
        //打印服务器列表
        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 {

        DistributedClient client = new DistributedClient();
        //获取ZooKeeper客户端连接
        client.getConnect();
        //获取servers子节点信息并监听,从中获取服务器的信息列表
        client.getServerList();
        //业务线程启动
        client.handleBussiness();

    }
}