《ZooKeeper》 Chapter 3 Getting Started with the ZooKeeper API

一、引入Java版驱动程序库依赖

如果使用Maven，在“pom.xml”文件中增加如下内容，引入对Java版驱动程序库的依赖。

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.8</version>
</dependency>

二、会话

关于“会话”的一些困惑解答参见《ZooKeeper中“会话”的那些事儿》。
在Java版驱动程序库中，表示“会话”的类是org.apache.zookeeper.ZooKeeper，创建会话和结束会话的代码片段如下。
注意在创建一个ZooKeeper Session的过程中，会创建一个Daemon线程，该线程负责维护该ZooKeeper Session。

package org.apache.zookeeper.book;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Session实现Watcher的目的是为了让监视器回调方法能够访问Session中的状态变量
 */
public class Session implements Watcher {
    private static final Logger logger = LoggerFactory.getLogger(Session.class);

    /**
     * ZooKeeper集群的端口号和地址列表
     */
    private static final String hostPort = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";

    /**
     * ZooKeeper Server判定会话超期的时间间隔
     */
    private static final Integer sessionTimeout = 60000;

    /**
     * ZooKeeper Session对象
     */
    private ZooKeeper zk;

    /**
     * ZooKeeper Session是否连接状态
     */
    private volatile boolean connected = false;

    /**
     * ZooKeeper Session是否过期状态
     */
    private volatile boolean expired = false;

    public Session() {}

    /**
     * 由单独的线程负责调用
     * 
     * @param watchedEvent
     */
    @Override
    public void process(WatchedEvent watchedEvent) {
        if (watchedEvent.getType() == Event.EventType.None) {
            if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                logger.info("session is created");

                connected = true;
            } else if (watchedEvent.getState() == Event.KeeperState.Disconnected) {
                logger.info("session is disconnected");

                connected = false;
            } else if (watchedEvent.getState() == Event.KeeperState.Expired) {
                logger.info("session expired");

                expired = true;
                connected = false;
            }
        }
    }

    /**
     * 创建一个ZooKeeper Session
     * 
     * @throws IOException
     */
    public void start() throws IOException {
        // 创建一个“Session”实例
        // hostPort：表示该“Session”实例关联的ZooKeeper集群的端口号和地址列表
        // sessionTimeout：表示该“Session”实例的过期时间。具体含义是，该“Session”实例关联到的ZooKeeper
        // Server在“sessionTimeout”时间内没有收到该“Session”实例关联到的ZooKeeper
        // Client的数据，该“Session”实例关联到的ZooKeeper Server则认为该“Session”实例过期
        // watcher：在该“Session”实例上设置的监视器，可以监听该“Session”实例上发生的事件
        zk = new ZooKeeper(hostPort, sessionTimeout, this);

        // 以上只是发出创建“Session”命令，“Session”完全建立需要一点时间
        while (!isConnected()) {
            try {
                Thread.sleep(100);
            } catch (Exception e) {
                logger.error("", e);
            }
        }
    }

    /**
     * 结束“Session”实例
     * 
     * @throws InterruptedException
     */
    public void stop() throws InterruptedException {
        zk.close();
    }

    public boolean isConnected() {
        return connected;
    }

    public void setConnected(boolean connected) {
        this.connected = connected;
    }

    public boolean isExpired() {
        return expired;
    }

    public void setExpired(boolean expired) {
        this.expired = expired;
    }

    public ZooKeeper getZk() {
        return zk;
    }

    public static void main(String[] args) throws IOException {
        Session session = new Session();
        session.start();
    }
}

三、监视器

监视器的基础接口是org.apache.zookeeper.Watcher，“二、会话”中的代码不仅介绍了“会话”实例，也介绍了“监视器”实例。
运行以上代码的结果部分截图如图1所示，这表明“会话”实例创建后，发生了一个事件，监视器实例监听到了这个事件，并在public void process(WatchedEvent watchedEvent)方法中将其内容打印了出来。

图1

四、实现一个简单的遵循“Master-Worker”架构体系的分布式应用程序

使用“Java版驱动程序库提供的API”编写应用程序有以下几点注意事项：
1、接口一般分为同步接口和异步接口，采用异步接口，编码比较简单
2、需要特殊处理ConnectionLossException异常，因为它不像NodeExistsException，NoNodeException等异常，对应的是确定的系统状态。在异步结果返回中，这些异常相对应的结果码分别为CONNECTIONLOSS，NODEEXISTS和NONODE

4.1、实现Master程序

4.1.1、Master程序业务逻辑

实现Master程序，有以下几点注意事项：
1、整个分布式应用程序中只有1个Master进程，利用ZooKeeper提供的服务，可以设定成功创建“/master”Znode节点的进程作为Master进程
2、整个分布式应用程序虽然只有一个Master进程，但还是需要多个候选Master进程，这样才能在原Master进程不能正常工作情况下，选举新的Master进程
3、原Master进程不能正常工作之时，对应的“/master”Znode节点应该自动删除，因此“/master”Znode节点的类型应为“Ephemeral ”，这样才能让候选Master进程能够被选举成为新的Master进程

4.1.2、Master程序实现中用到的接口说明

4.1.2.1、create异步接口

方法签名如下：

void create(String path,byte[] data,List<ACL> acl,CreateMode createMode,AsyncCallback.StringCallback cb,Object ctx)

方法签名含义介绍如下：
“path”：欲创建Znode节点的路径
“data”：欲创建Znode节点的内容
“acl”：欲创建Znode节点的访问权限控制设置，一般取为“Ids.OPEN_ACL_UNSAFE”
“createMode”：欲创建Znode节点的类型，有“Persistent，Ephemeral，Sequential”
“cb”：表示包含回调方法的回调实例
“ctx”：传递给回调方法的对象，可作为本方法与回调方法的通信数据对象

4.1.2.2、AsyncCallback.StringCallback类

AsyncCallback.StringCallback类定义如下：

public interface StringCallback extends AsyncCallback {
    void processResult(int rc, String path, Object ctx, String name);
}

“processResult”方法签名含义介绍如下：
“rc”：返回结果代码
“path”：调用“create”方法时，提供的Znode节点路径
“ctx”：调用“create”方法时，提供的“Object ctx”
“name”：欲创建Znode节点的名字，一般跟“path”一致，但是当欲创建Znode节点类型为“Sequential”时，就不一致了

4.1.2.3、getData异步接口

方法签名如下：

public void getData(String path, boolean watch, AsyncCallback.DataCallback cb, Object ctx)

方法签名含义介绍如下：
“path”：欲读取数据的Znode节点路径
“watch”：是否设置监视器，如果是“true”，那么使用的监视器就是在“ZooKeeper Session”上设置的监视器
“cb”：表示包含回调方法的回调实例
“ctx”：传递给回调方法的对象，可作为本方法与回调方法的通信数据对象

4.1.2.4、AsyncCallback.DataCallback类

AsyncCallback.DataCallback类定义如下：

public interface DataCallback extends AsyncCallback {
    void processResult(int rc, String path, Object ctx, byte[] data, Stat stat);
}

“processResult”方法签名含义介绍如下：
“rc”：返回结果代码
“path”：调用“getData”方法时，提供的Znode节点路径
“ctx”：调用“getData”方法时，提供的“Object ctx”
“data”：返回的Znode节点的内容，以字节数组形式表示
“stat”：使用Znode节点的元数据进行填充的对象

4.1.3、具体实现代码

package org.apache.zookeeper.book;

import java.io.IOException;
import java.util.Random;

import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.KeeperException.Code;

import org.apache.zookeeper.data.Stat;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Master程序
 */
public class Master {
    private static final Logger logger = LoggerFactory.getLogger(Master.class);

    /**
     * 需要的“Session”对象
     */
    private Session session;

    /**
     * Master进程状态
     */
    private volatile MasterStates state = MasterStates.RUNNING;

    private Random random = new Random(this.hashCode());

    /**
     * 作为Znode节点内容
     */
    private String serverId = Integer.toHexString(random.nextInt());

    public Master(Session session) {
        this.session = session;
    }

    /**
     * 创建“Session”
     * 
     * @throws IOException
     */
    void startZK() throws IOException {
        session.start();
    }

    /**
     * 尝试申请作为Master进程
     */
    synchronized public void runForMaster() {
        logger.info("Running for master");

        session.getZk().create("/master", serverId.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                masterCreateCallback, null);
    }

    /**
     * 申请创建"/master"Znode节点回调类实例
     */
    StringCallback masterCreateCallback = new StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();

                    break;
                case OK:
                    state = MasterStates.ELECTED;

                    // 其他处理，由于这里只是作为示例，因而不列出这个复杂的“其他处理”
                    // takeLeadership();

                    break;
                case NODEEXISTS:
                    state = MasterStates.NOTELECTED;

                    masterExists();

                    break;
                default:
                    state = MasterStates.NOTELECTED;

                    logger.error("Something went wrong when running for master.",
                            KeeperException.create(Code.get(rc), path));
            }

            logger.info("I‘m " + (state == MasterStates.ELECTED ? "" : "not ") + "the leader " + serverId);
        }
    };

    /**
     * 查看"/master"Znode是否真的存在
     */
    void checkMaster() {
        session.getZk().getData("/master", false, masterCheckCallback, null);
    }

    /**
     * 查看"/master"Znode节点是否真的存在的回调实例：<br/>
     * 1、如果连接中断，那么循环查看<br/>
     * 2、如果不存在"/master"Znode节点，那么本进程申请成为Master进程<br/>
     * 3、如果成功查询，那么判断是否是本进程<br/>
     */
    DataCallback masterCheckCallback = new DataCallback() {
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();

                    break;
                case NONODE:
                    runForMaster();

                    break;
                case OK:
                    if (serverId.equals(new String(data))) {
                        state = MasterStates.ELECTED;

                        // 其他处理，由于这里只是作为示例，因而不列出这个复杂的“其他处理”
                        // takeLeadership();
                    } else {
                        state = MasterStates.NOTELECTED;
                        masterExists();
                    }

                    break;
                default:
                    logger.error("Error when reading data.", KeeperException.create(Code.get(rc), path));
            }
        }
    };

    /**
     * 确认“/master”Znode节点是否真的存在
     */
    void masterExists() {
        session.getZk().exists("/master", masterExistsWatcher, masterExistsCallback, null);
    }

    StatCallback masterExistsCallback = new StatCallback() {
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    masterExists();

                    break;
                case OK:
                    // 的确真的存在，不可能是自身进程，因为对于同一个候选Master进程，只会启动一次
                    break;
                case NONODE:
                    state = MasterStates.RUNNING;

                    // 已经不存在
                    runForMaster();

                    logger.info("It sounds like the previous master is gone, " + "so let‘s run for master again.");

                    break;
                default:
                    checkMaster();
                    break;
            }
        }
    };

    /**
     * 监视器实例
     */
    Watcher masterExistsWatcher = new Watcher() {
        public void process(WatchedEvent e) {
            if (e.getType() == EventType.NodeDeleted) {
                runForMaster();
            }
        }
    };

    MasterStates getState() {
        return state;
    }

    /**
     * 关闭“Session”
     *
     * @throws InterruptedException
     */
    void stopZK() throws InterruptedException {
        session.stop();
    }

    public static void main(String args[]) throws Exception {
        Master m = new Master(new Session());
        m.startZK();

        m.runForMaster();

        Thread.sleep(10000);

        m.stopZK();
    }
}


enum MasterStates {
    RUNNING, ELECTED, NOTELECTED
};

4.2、实现Worker程序

4.2.1、主体实现代码

package org.apache.zookeeper.book;

import java.io.IOException;
import java.util.Random;

import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.KeeperException.Code;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Worker {
    private static final Logger logger = LoggerFactory.getLogger(Worker.class);

    /**
     * 需要的“Session”对象
     */
    private Session session;

    /**
     * Znode节点路径的一部分
     */
    private String serverId = Integer.toHexString((new Random()).nextInt());

    public Worker(Session session) {
        this.session = session;
    }

    public void startZK() throws IOException {
        this.session.start();
    }

    /**
     * 注册一个“Worker”
     */
    public void register() {
        String name = "worker-" + serverId;
        this.session.getZk().create("/workers/" + name, "Idle".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
                createWorkerCallback, null);
    }

    StringCallback createWorkerCallback = new StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (Code.get(rc)) {
                case CONNECTIONLOSS:
                    register();

                    break;
                case OK:
                    logger.info("Registered successfully: " + serverId);

                    break;
                case NODEEXISTS:
                    logger.warn("Already registered: " + serverId);

                    break;
                default:
                    logger.error("Something went wrong: ", KeeperException.create(Code.get(rc), path));
            }
        }
    };

    public void close() throws IOException {
        logger.info("Closing");
        try {
            this.session.stop();
        } catch (InterruptedException e) {
            logger.warn("ZooKeeper interrupted while closing");
        }
    }

    public static void main(String args[]) throws Exception {
        Worker w = new Worker(new Session());

        w.startZK();

        w.register();

        Thread.sleep(10000);

        w.close();
    }
}

4.2.2、重要实现代码

以下代码片段，在Worker程序中用于更改状态。

    String status;

    String name;

    AsyncCallback.StatCallback statusUpdateCallback = new AsyncCallback.StatCallback() {
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    updateStatus((String) ctx);
                    return;
            }
        }
    };

    synchronized private void updateStatus(String status) {
        if (status.equals(this.status)) {
            this.session.getZk().setData("/workers/" + name, status.getBytes(), -1, statusUpdateCallback, status);
        }
    }

    public void setStatus(String status) {
        this.status = status;
        updateStatus(status);
    }

对于以上代码片段，有两点需要特别注意。
1、status.equals(this.status)逻辑
出现以上逻辑主要是为了应对以下情形：主线程调用“setStatus”方法欲将状态设为“A”，由于ConnectionLossException异常未成功，执行回调方法的线程等待执行“updateStatus”方法；在这过程中，主线程又调用“setStatus”方法欲将状态设为“B”；此时，执行回调方法的线程执行“updateStatus”方法，欲将状态设为“A”。最后的状态本应为“B”，现为“A”，以上逻辑可以避免这种情况
2、updateStatus方法的synchronized关键词
分析可知，执行回调方法的线程和主线程都会尝试执行“updateStatus”方法，如果“updateStatus”方法中只是简单的“setData”方法，那么不会产生并发问题，而现在有status.equals(this.status)逻辑，需要通过使用synchronized关键词来达到线程安全的目的

五、线程相关

需要注意的是，主线程调用异步方法后立即返回，它其实只是发出了一个命令，另外会开启一个线程将发出请求的请求数据传输给“ZooKeeper Server”，对于调用异步方法而后续执行的回调方法也会由独立的线程进行执行。而且，为了保证顺序，只会有一个专门的线程用于执行该类回调方法，并不是该类中每个回调方法的执行都独立开启一个线程。
另外还有一类回调方法，即监视器实例中的处理方法，对于这类回调方法的线程相关情况，会在接下来的篇幅中进行介绍。

参考文献：
[1]https://github.com/fpj/zookeeper-book-example/tree/master/src/main/java/org/apache/zookeeper/book
[2]https://github.com/dslztx/zookeeper-book-example