public class RegistrationCenter {
    // 連接信息
    private String connectString = "192.168.10.11:2181,192.168.10.11:2182,192.168.10.11:2183";
    // 超時(shí)時(shí)間
    private int sessionTimeOut = 30000;
    private final String ROOT_PATH = "/servers";
    private ZooKeeper client;
    public RegistrationCenter() {
        this(null);
    }
    public RegistrationCenter(Consumer<List<String>> consumer) {
        try {
            getConnection(null == consumer ? null : watchedEvent -> {
                //監(jiān)聽服務(wù)器地址的上下線
                if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
                    try {
                        consumer.accept(subServers());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
            Stat stat = client.exists(ROOT_PATH, false);
            if (stat == null) {
                //創(chuàng)建根節(jié)點(diǎn)
                client.create(ROOT_PATH, ROOT_PATH.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * @param serverName 將服務(wù)器注冊(cè)到zk集群時(shí)，所需的服務(wù)名稱
     * @param metadata   服務(wù)元數(shù)據(jù)
     * @throws Exception
     */
    public void doRegister(String serverName, Metadata metadata) throws Exception {
        /**
         * ZooDefs.Ids.OPEN_ACL_UNSAFE: 此權(quán)限表示允許所有人訪問(wèn)該節(jié)點(diǎn)（服務(wù)器）
         * CreateMode.EPHEMERAL_SEQUENTIAL: 由于服務(wù)器是動(dòng)態(tài)上下線的，上線后存在，下線后不存在，所以是臨時(shí)節(jié)點(diǎn)
         * 而服務(wù)器一般都是有序號(hào)的，所以是臨時(shí)、有序的節(jié)點(diǎn).
         */
        String node = client.create(ROOT_PATH + "/" + serverName, metadata.toString().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(serverName + " 已經(jīng)上線");
    }
    /**
     * 發(fā)現(xiàn)/訂閱服務(wù)
     */
    public List<String> subServers() throws InterruptedException, KeeperException {
        List<String> zkChildren = client.getChildren(ROOT_PATH, true);
        List<String> servers = new ArrayList<>();
        zkChildren.forEach(node -> {
            //拼接服務(wù)完整信息
            try {
                byte[] data = client.getData(ROOT_PATH + "/" + node, false, null);
                servers.add(new String(data));
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        return servers;
    }
    private void getConnection(Watcher watcher) throws IOException {
        this.client = new ZooKeeper(connectString, sessionTimeOut, watcher);
    }
    /**
     * 服務(wù)元數(shù)據(jù)
     */
    public static class Metadata {
        public Metadata() {
        }
        public Metadata(String ip, int port) {
            this.ip = ip;
            this.port = port;
        }
        private String ip;
        private int port;
        public String getIp() {
            return ip;
        }
        public void setIp(String ip) {
            this.ip = ip;
        }
        public int getPort() {
            return port;
        }
        public void setPort(int port) {
            this.port = port;
        }
        @Override
        public String toString() {
            return "{" + "ip='" + ip + '\'' + ", port=" + port + '}';
        }
    }
}

public class ProviderServer {
    public static void main(String[] args) throws Exception {
        RegistrationCenter registrationCenter = new RegistrationCenter();
        registrationCenter.doRegister("provider", new RegistrationCenter.Metadata("127.0.0.1", 8080));
        Thread.sleep(Long.MAX_VALUE);
    }
}

public class ConsumerServer {
    public static void main(String[] args) throws Exception {
        RegistrationCenter registrationCenter = new RegistrationCenter(newServers -> {
            System.out.println("服務(wù)更新了..."+newServers);
        });
        List<String> servers = registrationCenter.subServers();
        System.out.println(servers);
        Thread.sleep(Long.MAX_VALUE);
    }
}