基于Zookeeper实现服务注册和服务发现功能

public class RegistrationCenter {
    // 连接信息
    private String connectString = "192.168.10.11:2181,192.168.10.11:2182,192.168.10.11:2183";
    // 超时时间
    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 -> {
                //监听服务器地址的上下线
                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) {
                //创建根节点
                client.create(ROOT_PATH, ROOT_PATH.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * @param serverName 将服务器注册到zk集群时，所需的服务名称
     * @param metadata   服务元数据
     * @throws Exception
     */
    public void doRegister(String serverName, Metadata metadata) throws Exception {
        /**
         * ZooDefs.Ids.OPEN_ACL_UNSAFE: 此权限表示允许所有人访问该节点（服务器）
         * CreateMode.EPHEMERAL_SEQUENTIAL: 由于服务器是动态上下线的，上线后存在，下线后不存在，所以是临时节点
         * 而服务器一般都是有序号的，所以是临时、有序的节点.
         */
        String node = client.create(ROOT_PATH + "/" + serverName, metadata.toString().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(serverName + " 已经上线");
    }
    /**
     * 发现/订阅服务
     */
    public List<String> subServers() throws InterruptedException, KeeperException {
        List<String> zkChildren = client.getChildren(ROOT_PATH, true);
        List<String> servers = new ArrayList<>();
        zkChildren.forEach(node -> {
            //拼接服务完整信息
            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);
    }
    /**
     * 服务元数据
     */
    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("服务更新了..."+newServers);
        });
        List<String> servers = registrationCenter.subServers();
        System.out.println(servers);
        Thread.sleep(Long.MAX_VALUE);
    }
}