ZooKeeper-API 監聽

以服務動態上下線通知為例

 

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<String> children = zkClient.getChildren("/servers", true);
        // 存儲服務器節點主機名稱集合
        ArrayList<String> 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 + "：上線");
    }
}

 

測試

1.直接運行 Client

2.運行 Server 后再查看 Client 的控制台

3.關閉  Server 后再查看 Client 的控制台

 

---

http://zookeeper.apache.org/doc/r3.4.14/javaExample.html

https://my.oschina.net/u/164027/blog/1921308