ZooKeeper-API 监听

2021-06-06 23:07

阅读：394

标签：reg inf 业务查看内容 osc info lis exists

以服务动态上下线通知为例

Client 监听服务器状态

public class DistributeClient {

    private String connectString = "127.0.0.1:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        BasicConfigurator.configure();
        DistributeClient client = new DistributeClient();
        // 获取 zookeeper 集群连接
        client.getConnect();
        // 注册监听
        client.getChlidren();

        Thread.sleep(Long.MAX_VALUE);
    }

    private void getConnect() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                try {
                    // 具体监听业务
                    getChlidren();
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    private void getChlidren() throws KeeperException, InterruptedException {
        if (zkClient.exists("/servers", false) == null) {
            zkClient.create("/servers", "server".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        // 监听 /servers 节点
        List children = zkClient.getChildren("/servers", true);
        // 存储服务器节点主机名称集合
        ArrayList hosts = new ArrayList();
        for (String child : children) {
            // 获取节点内容，即主机名称
            byte[] data = zkClient.getData("/servers/" + child, false, null);
            hosts.add(new String(data));
        }
        System.out.println("在线主机：" + hosts);
    }
}

Server 服务器，上线后 Client 端会收到通知

public class DistributeServer {

    private String connectString = "127.0.0.1:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;

    public static void main(String[] args) throws Exception {
        BasicConfigurator.configure();
        DistributeServer server = new DistributeServer();
        // 连接 zookeeper 集群
        server.getConnect();
        // 注册节点
        server.regist(UUID.randomUUID().toString());
        Thread.sleep(Long.MAX_VALUE);
    }

    private void getConnect() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // TODO Auto-generated method stub
            }
        });
    }

    private void regist(String hostname) throws KeeperException, InterruptedException {
        // 创建临时带序号节点
        zkClient.create("/servers/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(hostname + "：上线");
    }
}