Zookeeper如何实现分布式服务配置中心详解

Zookeeper如何实现分布式服务配置中心详解

什么是Zookeeper

Zookeeper是一个典型的分布式数据一致性解决方案，是Google Chubby在开源领域的实现，提供了分布式应用系统的协调服务，如配置维护、命名服务、同步服务、组服务等。

Zookeeper作为服务配置中心的应用

服务配置中心是比较常用的分布式架构中的一部分，它的目的是帮助我们实现应用程序的配置管理。Zookeeper正是通过其节点、事件机制等特性来支持分布式服务的协调和应用配置的管理。

Zookeeper实现分布式服务配置中心

配置项的注册和发现

通过节点机制，Zookeeper实现了配置项的注册和发现。我们可以将配置项写入到Zookeeper的特定节点下，通过监控该节点路径实现节点值变化的监听，使配置变化得到及时的通知。应用程序在启动时可以对Zookeeper的节点路径进行监听，当Zookeeper节点的值变化时，可以及时获取变化的值。

下面通过Java程序示例说明：

• 注册配置项

public class ConfigCenter {
    private ZooKeeper zooKeeper;
    private static String CONFIG_ROOT_NODE_PATH = "/config-center";
    private String configNodePath;

    public ConfigCenter(String connectString, int sessionTimeout) throws IOException, KeeperException, InterruptedException {
        this.zooKeeper = new ZooKeeper(connectString, sessionTimeout, null);
        if (zooKeeper.exists(CONFIG_ROOT_NODE_PATH, false) == null) {
            zooKeeper.create(CONFIG_ROOT_NODE_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    /**
     * 注册配置项
     * @param configName 配置项名称
     * @param configValue 配置项值
     * @return 配置项在Zookeeper中的节点路径
     * @throws KeeperException
     * @throws InterruptedException
     */
    public String registerConfig(String configName, String configValue) throws KeeperException, InterruptedException {
        configNodePath = zooKeeper.create(CONFIG_ROOT_NODE_PATH + "/" + configName, configValue.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        return configNodePath;
    }
}

• 监听配置项变化

public class ConfigCenterWatcher implements Watcher {
    private ZooKeeper zooKeeper;
    private String configNodePath;

    public ConfigCenterWatcher(ZooKeeper zooKeeper, String configNodePath) {
        this.zooKeeper = zooKeeper;
        this.configNodePath = configNodePath;
    }

    @Override
    public void process(WatchedEvent event) {
        switch (event.getType()) {
            case NodeDataChanged:
                try {
                    byte[] data = zooKeeper.getData(configNodePath, this, null);
                    System.out.println("配置项已经更新，新值为：" + new String(data));
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
                break;
            default:
                break;
        }
    }
}

• 使用示例

public class ConfigCenterTest {
    private static String CONNECT_STRING = "localhost:2181";
    private static int SESSION_TIMEOUT = 10000;
    private static String CONFIG_NAME = "testConfig";
    private static String CONFIG_VALUE = "testConfigValue";

    @Test
    public void testConfigCenter() throws IOException, KeeperException, InterruptedException, InterruptedException {
        ConfigCenter configCenter = new ConfigCenter(CONNECT_STRING, SESSION_TIMEOUT);
        String configNodePath = configCenter.registerConfig(CONFIG_NAME, CONFIG_VALUE);

        ZooKeeper zooKeeper = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, new ConfigCenterWatcher(zooKeeper, configNodePath));
        byte[] data = zooKeeper.getData(configNodePath, true, null);
        System.out.println("配置项初始值为：" + new String(data));

        Thread.sleep(Long.MAX_VALUE);
    }
}

该示例通过Zookeeper实现了配置项注册和变化时的通知。

分布式应用系统中的服务发现

Zookeeper可以帮助我们实现分布式应用系统中的服务发现。在分布式系统中，服务可用性监控和负载均衡都是必须的特性。Zookeeper作为注册中心能够协调分布式应用系统中的服务注册和发现，也就是说，实现了单台服务器到集群间的扩展，让分布式服务可以向Zookeeper注册自己的服务地址和相关信息，也可以从Zookeeper查询服务注册信息。

下面通过Java程序示例说明：

• 服务注册

public class ServerRegistry {
    private ZooKeeper zooKeeper;
    private static String SERVER_NODE_PATH = "/server";

    public ServerRegistry(String connectString, int sessionTimeout) throws IOException, KeeperException, InterruptedException {
        this.zooKeeper = new ZooKeeper(connectString, sessionTimeout, null);
        if (zooKeeper.exists(SERVER_NODE_PATH, false) == null) {
            zooKeeper.create(SERVER_NODE_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    /**
     * 服务注册
     * @param serverName 服务名
     * @param serverAddress 服务地址
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void registerServer(String serverName, String serverAddress) throws KeeperException, InterruptedException {
        String serverNodePath = SERVER_NODE_PATH + "/" + serverName;
        if (zooKeeper.exists(serverNodePath, false) == null) {
            zooKeeper.create(serverNodePath, serverAddress.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(serverName + "已注册，地址为：" + serverAddress);
        }
    }
}

• 服务发现

public class LoadBalance {
    private ZooKeeper zooKeeper;
    private static String SERVER_NODE_PATH = "/server";
    private String serviceName;

    public LoadBalance(String connectString, int sessionTimeout, String serviceName) throws IOException, KeeperException, InterruptedException {
        this.zooKeeper = new ZooKeeper(connectString, sessionTimeout, null);
        if (zooKeeper.exists(SERVER_NODE_PATH, false) == null) {
            throw new RuntimeException("no server to use");
        }
        this.serviceName = serviceName;
    }

    /**
     * 选择服务器
     * @return 服务器地址
     * @throws KeeperException
     * @throws InterruptedException
     */
    public String selectServer() throws KeeperException, InterruptedException {
        String serverNodePath = SERVER_NODE_PATH + "/" + serviceName;
        List<String> servers = zooKeeper.getChildren(serverNodePath, true);
        String serverAddress = null;
        if (servers != null && servers.size() > 0) {
            if (servers.size() == 1) {
                serverAddress = new String(zooKeeper.getData(serverNodePath + "/" + servers.get(0), true, null));
                System.out.println("唯一服务器地址为：" + serverAddress);
            } else {
                int index = ThreadLocalRandom.current().nextInt(servers.size());
                serverAddress = new String(zooKeeper.getData(serverNodePath + "/" + servers.get(index), true, null));
                System.out.println("随机选择的服务器地址为：" + serverAddress);
            }
        }
        return serverAddress;
    }
}

• 使用示例

public class ServerRegistryTest {
    private static String CONNECT_STRING = "localhost:2181";
    private static int SESSION_TIMEOUT = 10000;
    private static String SERVER_NAME = "testServer";
    private static String SERVER_ADDRESS_1 = "192.168.1.100:8080";
    private static String SERVER_ADDRESS_2 = "192.168.1.101:8080";
    private static String SERVICE_NAME = "testService";

    @Test
    public void testServerRegistry() throws IOException, KeeperException, InterruptedException {
        ServerRegistry serverRegistry = new ServerRegistry(CONNECT_STRING, SESSION_TIMEOUT);
        serverRegistry.registerServer(SERVER_NAME, SERVER_ADDRESS_1);
        serverRegistry.registerServer(SERVER_NAME, SERVER_ADDRESS_2);

        Thread.sleep(Long.MAX_VALUE);
    }

    @Test
    public void testLoadBalance() throws IOException, KeeperException, InterruptedException {
        LoadBalance loadBalance = new LoadBalance(CONNECT_STRING, SESSION_TIMEOUT, SERVICE_NAME);
        while (true) {
            try {
                String serverAddress = loadBalance.selectServer();
                System.out.println(serverAddress);
            } catch (Exception e) {
                System.out.println("no server to use");
                Thread.sleep(2000);
            }
        }
    }
}

这个示例通过Zookeeper实现了服务注册和发现，达到了负载均衡的效果。