配置服务 Configuration Service

一个基本的ZooKeeper实现的服务就是“配置服务”，集群中的服务器可以通过ZooKeeper共享一个通用的配置数据。从表面上，ZooKeeper可以理解为一个配置数据的高可用存储服务，为应用提供检索和更新配置数据服务。我们可以使用ZooKeeper的观察模式实现一个活动的配置服务，当配置数据发生变化时，可以通知与配置相关客户端。

接下来，我们来实现一个这样的活动配置服务。首先，我们设计用znode来存储key-value对，我们在znode中存储一个String类型的数据作为value，用znode的path来表示key。然后，我们实现一个client，这个client可以在任何时候对数据进行跟新操作。那么这个设计的ZooKeeper数据模型应该是：master来更新数据，其他的worker也随之将数据更新，就像HDFS的namenode那样。

我们在一个叫做ActiveKeyValueStore的类中编写代码如下：

  1. public class ActiveKeyValueStore extends ConnectionWatcher {
  3.   private static final Charset CHARSET = Charset.forName("UTF-8");
  5.   public void write(String path, String value) throws InterruptedException,
  6.       KeeperException {
  7.     Stat stat = zk.exists(path, false);
  8.     if (stat == null) {
  9.       zk.create(path, value.getBytes(CHARSET), Ids.OPEN_ACL_UNSAFE,
  10.           CreateMode.PERSISTENT);
  11.     } else {
  12.       zk.setData(path, value.getBytes(CHARSET), -1);
  13.     }
  14.   }
  15. }

write()方法主要实现将给定的key-value对写入到ZooKeeper中。这其中隐含了创建一个新的znode和更新一个已存在的znode的实现方法的不同。那么操作之前，我们需要根据exists()来判断znode是否存在，然后再根据情况进行相关的操作。其他值得一提的就是String类型的数据在转换成byte[]时，使用的字符集是UTF-8。

我们为了说明ActiveKeyValueStore怎么使用，我们考虑实现一个ConfigUpdater类来实现更新配置。下面代码实现了一个在一些随机时刻更新配置数据的应用。

  1. public class ConfigUpdater {
  3.   public static final String PATH = "/config";
  5.   private ActiveKeyValueStore store;
  6.   private Random random = new Random();
  8.   public ConfigUpdater(String hosts) throws IOException, InterruptedException {
  9.     store = new ActiveKeyValueStore();
  10.     store.connect(hosts);
  11.   }
  13.   public void run() throws InterruptedException, KeeperException {
  14.     while (true) {
  15.       String value = random.nextInt(100) + "";
  16.       store.write(PATH, value);
  17.       System.out.printf("Set %s to %s\n", PATH, value);
  18.       TimeUnit.SECONDS.sleep(random.nextInt(10));
  19.     }
  20.   }
  22.   public static void main(String[] args) throws Exception {
  23.     ConfigUpdater configUpdater = new ConfigUpdater(args[0]);
  24.     configUpdater.run();
  25.   }
  26. }

上面的代码很简单。在ConfigUpdater的构造函数中，ActiveKeyValueStore对象连接到ZooKeeper服务。然后run()不断的循环运行，使用一个随机数不断的随机更新/configznode上的值。

下面我们来看一下，如何读取/config上的值。首先，我们在ActiveKeyValueStore中实现一个读方法。

  1. public String read(String path, Watcher watcher) throws InterruptedException,
  2.       KeeperException {
  3.     byte[] data = zk.getData(path, watcher, null/*stat*/);
  4.     return new String(data, CHARSET);
  5.   }

ZooKeeper的getData()方法的参数包含：path，一个Watcher对象和一个Stat对象。Stat对象中含有从getData()返回的值，并且负责接收回调信息。这种方式下，调用者不仅可以获得数据，还能够获得znode的metadata。

做为服务的consumer，ConfigWatcher以观察者身份，创建一个ActiveKeyValueStore对象，并且在启动以后调用read()函数（在dispalayConfig()函数中）获得相关数据。

下面的代码实现了一个以观察模式获得ZooKeeper中的数据更新的应用，并将值到后台中。

  1. public class ConfigWatcher implements Watcher {
  3.   private ActiveKeyValueStore store;
  5.   public ConfigWatcher(String hosts) throws IOException, InterruptedException {
  6.     store = new ActiveKeyValueStore();
  7.     store.connect(hosts);
  8.   }
  10.   public void displayConfig() throws InterruptedException, KeeperException {
  11.     String value = store.read(ConfigUpdater.PATH, this);
  12.     System.out.printf("Read %s as %s\n", ConfigUpdater.PATH, value);
  13.   }
  15.   @Override
  16.   public void process(WatchedEvent event) {
  17.     if (event.getType() == EventType.NodeDataChanged) {
  18.       try {
  19.         displayConfig();
  20.       } catch (InterruptedException e) {
  21.         System.err.println("Interrupted. Exiting.");        
  22.         Thread.currentThread().interrupt();
  23.       } catch (KeeperException e) {
  24.         System.err.printf("KeeperException: %s. Exiting.\n", e);        
  25.       }
  26.     }
  27.   }
  29.   public static void main(String[] args) throws Exception {
  30.     ConfigWatcher configWatcher = new ConfigWatcher(args[0]);
  31.     configWatcher.displayConfig();
  33.     // stay alive until process is killed or thread is interrupted
  34.     Thread.sleep(Long.MAX_VALUE);
  35.   }
  36. }

当ConfigUpadater更新znode时，ZooKeeper将触发一个EventType.NodeDataChanged的事件给观察者。ConfigWatcher将在他的process()函数中获得这个时间，并将显示读取到的最新的版本的配置数据。

由于观察模式的触发是一次性的，所以每次都要调用ActiveKeyValueStore的read()方法，这样才能获得未来的更新数据。我们不能确保一定能够接受到更新通知事件，因为在接受观察事件和下一次读取之间的窗口期内，znode可能被改变了（有可能很多次），但是client可能没有注册观察模式，所以client不会接到znode改变的通知。在配置服务中这不是一个什么问题，因为client只关心配置数据的最新版本。然而，建议读者关注一下这个潜在的问题。

让我们来看一下控制台打印的ConfigUpdater运行结果：

  1. % java ConfigUpdater localhost
  2. Set /config to 79
  3. Set /config to 14
  4. Set /config to 78

然后立即在另外的控制台终端窗口中运行ConfigWatcher:

  1. % java ConfigWatcher localhost
  2. Read /config as 79
  3. Read /config as 14
  4. Read /config as 78