博客

一、概述
这次案例主要是讲解使用Zookeeper集群实现服务器上线下线监听，从而实现一系列业务逻辑。

场景模拟：
创建三个服务器，分别为servers下的server1，server2，server3，使用一个客户端进行监听，监听三台服务器的节点状态的变化情况，当有服务器宕机或者节点数据发生变化时，客户端可以重新获取注册列表，从而重新获取服务器的数据。

案例图解如下图：

二、Java代码实现
【1】创建ZookeeperServer类，如下：

package zookeeperdemo;

import org.apache.zookeeper.*;

import java.io.IOException;

/**
 * @author czd
 */
public class ZookeeperServer {
    private static ZooKeeper zkClient = null;

    /**
     * 连接Zookeeper集群服务器端
     *
     * @return
     */
    public static ZooKeeper getInstance() {
        if (zkClient == null) {
            String connectPath = "192.168.223.128:2181,192.168.223.129:2181,192.168.223.131:2181";
            int sessionTimeout = 2000;
            try {
                zkClient = new ZooKeeper(connectPath, sessionTimeout, new Watcher() {
                    @Override
                    public void process(WatchedEvent watchedEvent) {

                    }
                });
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return zkClient;
    }

    /**
     * 创建服务器节点
     * ZooDefs.Ids.OPEN_ACL_UNSAFE:访问控制权限，这里是开放的，需不需要权限
     * CreateMode.EPHEMERAL_SEQUENTIAL：节点类型，有编码的短暂的节点
     * @param zooKeeper
     * @param content
     * @return
     */
    public static String createNode(ZooKeeper zooKeeper, String content) {
        String result = "";
        try {
            result = zooKeeper.create("/servers/server", content.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return result;
    }

    public static void main(String[] args) {
        ZooKeeper zooKeeper = getInstance();
        String result = createNode(zooKeeper, "Hello World！");
        System.out.println("result: " + result);

        //业务逻辑
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

【2】创建ZookeeperClient类，如下：

package zookeeperdemo;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author czd
 */
public class ZookeeperClient {
    private static ZooKeeper zkClient = null;

    /**
     * 连接Zookeeper集群客户端
     *
     * @return
     */
    public static ZooKeeper getInstance() {
        if (zkClient == null) {
            String connectPath = "192.168.223.128:2181,192.168.223.129:2181,192.168.223.131:2181";
            int sessionTimeout = 2000;
            try {
                zkClient = new ZooKeeper(connectPath, sessionTimeout, new Watcher() {
                    @Override
                    public void process(WatchedEvent watchedEvent) {
                        //当服务器状态发生变化时，在这里重新获取注册监听列表
                        System.out.println("---------start!----------");
                        register(zkClient);
                        System.out.println("---------end!----------");
                    }
                });
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return zkClient;
    }

    /**
     * 注册监听
     * @param zooKeeper
     * @return
     */
    public static List<String> register(ZooKeeper zooKeeper) {
        List<String> list = new ArrayList<>();
        try {
            List<String> children = zooKeeper.getChildren("/servers", true);
            for (String child : children) {
                //获取/servers/child 下的内容，例如获取/servers/server0000000000节点下的内容
                byte[] bytes = zooKeeper.getData("/servers/" + child, false, null);
                list.add(new String(bytes));
            }

            //打印当前注册监听到的节点的内容
            for (String str : list) {
                System.out.println(str);
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return list;
    }

    public static void main(String[] args) {
        ZooKeeper zooKeeper = getInstance();
        register(zooKeeper);

        //业务逻辑
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

结果如图所示：
①服务端

②客户端