服務發現：指對叢集中的服務上下線做統一管理，每個工作伺服器都可以作為資料的釋出方，向叢集註冊自己的基本資訊，而讓某些監控伺服器作為訂閱方，訂閱工作伺服器的基本資訊。當工作伺服器的基本資訊改變時，如服務上下線、伺服器的角色或服務範圍變更，那麼監控伺服器可以得到通知並響應這些變化。

實現程式碼如下：

import com.alibaba.fastjson.JSON;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;

 
/**
 * 代表工作伺服器
 */
public class WorkServer {

    private ZkClient zkClient;
    // ZooKeeper
    private String configPath;
    // ZooKeeper叢集中servers節點的路徑
    private String serversPath;
    // 當前工作伺服器的基本資訊
    private ServerData serverData;
    // 當前工作伺服器的配置資訊
    private ServerConfig serverConfig;
    private
 
 IZkDataListener dataListener;

    public WorkServer(String configPath, String serversPath,
                      ServerData serverData, ZkClient zkClient, ServerConfig initConfig) {
        this.zkClient = zkClient;
        this.serversPath = serversPath;
        this.configPath = configPath;
        
 
this.serverConfig = initConfig;
        this.serverData = serverData;

        this.dataListener = new IZkDataListener() {

            public void handleDataDeleted(String dataPath) throws Exception {

            }

            public void handleDataChange(String dataPath, Object data)
                    throws Exception {
                String retJson = new String((byte[])data);
                ServerConfig serverConfigLocal = (ServerConfig) JSON.parseObject(retJson,ServerConfig.class);
                updateConfig(serverConfigLocal);
                System.out.println("new Work server config is:"+serverConfig.toString());

            }
        };

    }

    // 啟動伺服器
    public void start() {
        System.out.println("work server start...");
        initRunning();
    }

    // 停止伺服器
    public void stop() {
        System.out.println("work server stop...");
        zkClient.unsubscribeDataChanges(configPath, dataListener); // 取消監聽config節點
    }

    // 伺服器初始化
    private void initRunning() {
        registMe(); // 註冊自己
        zkClient.subscribeDataChanges(configPath, dataListener); // 訂閱config節點的改變事件
    }

    // 啟動時向zookeeper註冊自己的註冊函式
    private void registMe() {
        String mePath = serversPath.concat("/").concat(serverData.getAddress());

        try {
            zkClient.createEphemeral(mePath, JSON.toJSONString(serverData)
                    .getBytes());
        } catch (ZkNoNodeException e) {
            zkClient.createPersistent(serversPath, true);
            registMe();
        }
    }

    // 更新自己的配置資訊
    private void updateConfig(ServerConfig serverConfig) {
        this.serverConfig = serverConfig;
    }

}

/**
 * 排程類
 */
public class SubscribeZkClient {

    private static final int  CLIENT_QTY = 5; // Work Server數量

    private static final String  ZOOKEEPER_SERVER = "192.168.1.105:2181";

    private static final String  CONFIG_PATH = "/config";
    private static final String  COMMAND_PATH = "/command";
    private static final String  SERVERS_PATH = "/servers";

    public static void main(String[] args) throws Exception {

        List<ZkClient> clients = new ArrayList<ZkClient>();
        List<WorkServer>  workServers = new ArrayList<WorkServer>();
        ManageServer manageServer = null;

        try {

            // 建立一個預設的配置
            ServerConfig initConfig = new ServerConfig();
            initConfig.setDbPwd("123456");
            initConfig.setDbUrl("jdbc:mysql://localhost:3306/mydb");
            initConfig.setDbUser("root");

            // 例項化一個Manage Server
            ZkClient clientManage = new ZkClient(ZOOKEEPER_SERVER, 5000, 5000, new BytesPushThroughSerializer());
            manageServer = new ManageServer(SERVERS_PATH, COMMAND_PATH,CONFIG_PATH,clientManage,initConfig);
            manageServer.start(); // 啟動Manage Server

            // 建立指定個數的工作伺服器
            for ( int i = 0; i < CLIENT_QTY; ++i ) {
                ZkClient client = new ZkClient(ZOOKEEPER_SERVER, 5000, 5000, new BytesPushThroughSerializer());
                clients.add(client);
                ServerData serverData = new ServerData();
                serverData.setId(i);
                serverData.setName("WorkServer#"+i);
                serverData.setAddress("192.168.1."+i);

                WorkServer  workServer = new WorkServer(CONFIG_PATH, SERVERS_PATH, serverData, client, initConfig);
                workServers.add(workServer);
                workServer.start();    // 啟動工作伺服器

            }

            System.out.println("敲回車鍵退出！\n");
            new BufferedReader(new InputStreamReader(System.in)).readLine();

        } finally {
            System.out.println("Shutting down...");

            for ( WorkServer workServer : workServers ) {
                try {
                    workServer.stop();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            for ( ZkClient client : clients ) {
                try {
                    client.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }
    }

}

/**
 * 伺服器基本資訊
 */
public class ServerData {

    private String address;
    private Integer id;
    private String name;

    public String getAddress() {
        return address;
    }
    public void setAddress(String address) {
        this.address = address;
    }
    public Integer getId() {
        return id;
    }
    public void setId(Integer id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "ServerData [address=" + address + ", id=" + id + ", name="
                + name + "]";
    }

}

/**
 * 配置資訊
 */
public class ServerConfig {

    private String dbUrl;
    private String dbPwd;
    private String dbUser;
    public String getDbUrl() {
        return dbUrl;
    }
    public void setDbUrl(String dbUrl) {
        this.dbUrl = dbUrl;
    }
    public String getDbPwd() {
        return dbPwd;
    }
    public void setDbPwd(String dbPwd) {
        this.dbPwd = dbPwd;
    }
    public String getDbUser() {
        return dbUser;
    }
    public void setDbUser(String dbUser) {
        this.dbUser = dbUser;
    }

    @Override
    public String toString() {
        return "ServerConfig [dbUrl=" + dbUrl + ", dbPwd=" + dbPwd
                + ", dbUser=" + dbUser + "]";
    }

}

import com.alibaba.fastjson.JSON;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import java.util.List;

public class ManageServer {

    // zookeeper的servers節點路徑
    private String serversPath;
    // zookeeper的command節點路徑
    private String commandPath;
    // zookeeper的config節點路徑
    private String configPath;
    private ZkClient zkClient;
    private ServerConfig config;
    // 用於監聽servers節點的子節點列表的變化
    private IZkChildListener childListener;
    // 用於監聽command節點資料內容的變化
    private IZkDataListener dataListener;
    // 工作伺服器的列表
    private List<String> workServerList;

    public ManageServer(String serversPath, String commandPath,
                        String configPath, ZkClient zkClient, ServerConfig config) {
        this.serversPath = serversPath;
        this.commandPath = commandPath;
        this.zkClient = zkClient;
        this.config = config;
        this.configPath = configPath;
        this.childListener = new IZkChildListener() {

            public void handleChildChange(String parentPath,
                                          List<String> currentChilds) throws Exception {
                // TODO Auto-generated method stub
                workServerList = currentChilds; // 更新記憶體中工作伺服器列表

                System.out.println("work server list changed, new list is ");
                execList();

            }
        };
        this.dataListener = new IZkDataListener() {

            public void handleDataDeleted(String dataPath) throws Exception {
                // TODO Auto-generated method stub
                // ignore;
            }

            public void handleDataChange(String dataPath, Object data)
                    throws Exception {
                // TODO Auto-generated method stub
                String cmd = new String((byte[]) data);
                System.out.println("cmd:"+cmd);
                exeCmd(cmd); // 執行命令

            }
        };

    }

    private void initRunning() {
        zkClient.subscribeDataChanges(commandPath, dataListener);
        zkClient.subscribeChildChanges(serversPath, childListener);
    }

    /*
     * 1: list 2: create 3: modify
     */
    private void exeCmd(String cmdType) {
        if ("list".equals(cmdType)) {
            execList();

        } else if ("create".equals(cmdType)) {
            execCreate();
        } else if ("modify".equals(cmdType)) {
            execModify();
        } else {
            System.out.println("error command!" + cmdType);
        }

    }

    // 列出工作伺服器列表
    private void execList() {
        System.out.println(workServerList.toString());
    }

    // 建立config節點
    private void execCreate() {
        if (!zkClient.exists(configPath)) {
            try {
                zkClient.createPersistent(configPath, JSON.toJSONString(config)
                        .getBytes());
            } catch (ZkNodeExistsException e) {
                zkClient.writeData(configPath, JSON.toJSONString(config)
                        .getBytes()); // config節點已經存在，則寫入內容就可以了
            } catch (ZkNoNodeException e) {
                String parentDir = configPath.substring(0,
                        configPath.lastIndexOf('/'));
                zkClient.createPersistent(parentDir, true);
                execCreate();
            }
        }
    }

    // 修改config節點內容
    private void execModify() {
        // 我們隨意修改config的一個屬性就可以了
        config.setDbUser(config.getDbUser() + "_modify");

        try {
            zkClient.writeData(configPath, JSON.toJSONString(config).getBytes());
        } catch (ZkNoNodeException e) {
            execCreate(); // 寫入時config節點還未存在，則建立它
        }
    }

    // 啟動工作伺服器
    public void start() {
        initRunning();
    }

    // 停止工作伺服器
    public void stop() {
        zkClient.unsubscribeChildChanges(serversPath, childListener);
        zkClient.unsubscribeDataChanges(commandPath, dataListener);
    }

}