拿董老師的案例：設計一個基於 Zookeeper 的 Hadoop 配置管理服務， 該服務由兩個功能模組組成：
1）  配置檔案更新元件 HadoopConfigUpdater （管理員修改配置檔案後，執行該元件，將資料寫入 Zookeeper） ，可將管理員修改的 Hadoop  conf 目錄打包成 conf.zip，並寫入 Zookeeper 中/conf 節點（znode）中；
2）  客戶端監聽者 HadoopConfigWatcher（執行在各個節點上） ，可監聽/conf 節點，一旦得知節點資料發生變化，則讀取 conf.zip，並解壓後替換所在節點的 hadoop conf 目錄。

下面的程式碼就是一個例子，比較簡單，不作說明了。如果要實現上面的場景需要修改ActiveKeyValueStore的write和read方法，使用ByteArrayOutputStream或ByteBuffer；

如果執行：

1、下載程式碼：https://github.com/sleberknight/zookeeper-samples/tree/master/src/main/java/com/nearinfinity/examples/zookeeper

2、下載zookeeper-3.4.*.tar.gz原始碼包，匯入jar包：log4j，slf4j-api，slf4j-log4j,zooKeeper-四個jar包。

import java.nio.charset.Charset;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import com.nearinfinity.examples.zookeeper.util.ConnectionWatcher;

public class ActiveKeyValueStore extends ConnectionWatcher {

    private static final Charset CHARSET = Charset.forName("UTF-8");

    public void write(String path, String value) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(path, false);
        if (stat == null) {
            zk.create(path, value.getBytes(CHARSET), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        else {
            zk.setData(path, value.getBytes(CHARSET), -1);
        }
    }

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

package com.nearinfinity.examples.zookeeper.confservice;



import java.io.IOException;

import java.util.Random;

import java.util.concurrent.TimeUnit;



import org.apache.zookeeper.KeeperException;



public class ConfigUpdater {



    public static final String PATH = "/config";



    private ActiveKeyValueStore _store;

    private Random _random = new Random();



    public ConfigUpdater(String hosts) throws IOException, InterruptedException {

        _store = new ActiveKeyValueStore();

        _store.connect(hosts);

    }



    public void run() throws InterruptedException, KeeperException {

        //noinspection InfiniteLoopStatement

        while (true) {

            String value = _random.nextInt(100) + "";

            _store.write(PATH, value);

            System.out.printf("Set %s to %s\n", PATH, value);

            TimeUnit.SECONDS.sleep(_random.nextInt(10));

        }

    }



    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

        ConfigUpdater updater = new ConfigUpdater("127.0.0.1:2183");

        updater.run();

    }



}

package com.nearinfinity.examples.zookeeper.confservice;

import java.io.IOException;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

public class ConfigWatcher implements Watcher {

    private ActiveKeyValueStore _store;

    public ConfigWatcher(String hosts) throws InterruptedException, IOException {
        _store = new ActiveKeyValueStore();
        _store.connect(hosts);
    }

    public void displayConfig() throws InterruptedException, KeeperException {
        String value = _store.read(ConfigUpdater.PATH, this);
        System.out.printf("Read %s as %s\n", ConfigUpdater.PATH, value);
    }


    @Override
    public void process(WatchedEvent event) {
        System.out.printf("Process incoming event: %s\n", event.toString());
        if (event.getType() == Event.EventType.NodeDataChanged) {
            try {
                displayConfig();
            }
            catch (InterruptedException e) {
                System.err.println("Interrupted. Exiting");
                Thread.currentThread().interrupt();
            }
            catch (KeeperException e) {
                System.err.printf("KeeperException: %s. Exiting.\n", e);
            }
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ConfigWatcher watcher = new ConfigWatcher(args[0]);
        watcher.displayConfig();

        Thread.sleep(Long.MAX_VALUE);
    }

}