ZooKeeper JAVA API 之环境准备和创建会话

    Zookeeper是一个开放源代码的分布式协调服务，由雅虎创建，是Google Chubby的开源实现。Zookeeper的设计目标是将那些复杂且容易出错的分布式一致性服务封装起来，构成一个高效可靠的元语集，并以一系列简单易用的接口提供给用户使用。

单机模式部署与运行（Windows）

1. 确保已经安装了JAVA 1.6及其以上版本的JDK
2. 下载Zookeeper    http://zookeeper.apache.org/releases.html  目前稳定版本即stable版本为3.4.6。将下载完成的压缩包解压到%ZK_HOME%下，其中ZK_HOME 目录可以随意设定，如：D://programes 等
3. 配置文件 zoo.cfg ,单机模式下，只需要将%ZK_HOME%/conf/zoo_sample.cfg文件重命名即可
4. 启动服务，在%ZK_HOME%bin目录下执行zkServer.cmd即可启动zookeeper服务器

Zookeeper 可执行脚本说明

1. zkCleanup  清理ZooKeeper历史数据，包括事务日志和快照文件
2. zkCli  ZooKeeper的一个建议客户端
3. zkEnv  设置ZooKeeper的环境变量
4. zkServer  ZooKeeper服务器的启动、停止和重启脚本

创建会话

       客户端可以通过创建一个ZooKeeper实例来连接ZooKeeper服务器。该类中的注释这样写道：

ZooKeeper类是ZooKeeper客户端库的主类。要使用ZooKeeper服务，应用程序首先必须要实现一个ZooKeeper的实例。所有的迭代都将通过调用ZooKeeper类的方法来完成。这个类的方法都是线程安全的除非另有说明。
 一旦一个与服务器的连接被建立，会分配给客户端一个Session ID. 客户端会周期性地发送心跳到服务器端，以保持会话有效。
 只要Session ID 有效，应用程序都可以通过客户端调用ZooKeeper的 API
  如果因为某些原因，导致客户端未能在指定的时间段内（超过SessionTimeout的值）发送心跳到服务器端，则该会话将过期，Session ID 将失效。
  如果客户端对象不可用，为了调用ZooKeeper API,应用程序必须创建一个新的客户端

 

ZooKeeper的构造方法有：

 ZooKeeper(String connectString,int sessionTimeout,Watcher watcher)

 ZooKeeper(String connectString,int sessionTimeout,Watcher watcher,boolean canBeReadOnly)

 ZooKeeper(String connectString,int sessionTimeout,Watcher watcher,long sessionId,boolean canBeReadOnly)

  ZooKeeper(String connectString,int sessionTimeout,Watcher watcher,long sessionId,byte[] sessionPasswd, boolean canBeReadOnly)

   connectString 是指ZooKeeper服务器的列表，host:prot,host:port 这样的格式构成，即主机:端口号通过英文逗号分隔。当然如果是单击模式下只需要一个;

    sessionTimeout 值会话的超时时间，单位是毫秒

  watcher Watcher事件的处理器，如果设置为null,表示不需要默认的Watcher事件处理器

   canBeReadOnly 用于表示当前会话是否支持 read-only

    sessionId,sessionPasswd 分别代表 会话的Id和密钥。这两个参数可以唯一确定一个会话。可以通过ZooKeeper 实例调用 getSesssionId()和getSessionPasswd()方法获取。

 

/**
 *创建一个最基本的ZooKeeper会话示例
 * @author zhangwei_david
 * @version $Id: ZKConstructorDemo.java, v 0.1 2015年5月1日 下午5:55:21 zhangwei_david Exp $
 */
public class ZKConstructorDemo implements Watcher {

    private static CountDownLatch connectedSemphore = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
        //创建Zookeeper会话实例
        ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new ZKConstructorDemo());
        // 输出当前会话的状态
        System.out.println("zk客户端的状态是：" + zookeeper.getState());
        System.out.println("zk 客户端的sessionId=" + zookeeper.getSessionId() + ",  sessionPasswd是："
                           + new String(zookeeper.getSessionPasswd()));
        try {
            // 当前闭锁在为0之前一直等待，除非线程中断
            connectedSemphore.await();
        } catch (Exception e) {
            System.out.println("Zookeeper session established");
        }
    }

    /**
     * @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.WatchedEvent)
     */
    public void process(WatchedEvent event) {
        System.out.println("Receive watched event:" + event);
        //如果客户端已经处于连接状态闭锁减去1
        if (KeeperState.SyncConnected == event.getState()) {
            connectedSemphore.countDown();
        }
    }
}

 结果：

2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:host.name=David
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:java.version=1.8.0
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:java.io.tmpdir=C:\Users\Lenovo\AppData\Local\Temp\
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:java.compiler=<NA>
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:os.name=Windows 8.1
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:os.arch=x86
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:os.version=6.3
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:user.name=Lenovo
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:user.home=C:\Users\Lenovo
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Client environment:user.dir=H:\Alipay.com\workspace4alipay\demo
2015-05-01 19:03:43  [ main:0 ] - [ INFO ]  Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=5000 watcher=com.cathy.demo.zk.ZKConstructorDemo@1e717c2
zk客户端的状态是：CONNECTING
zk 客户端的sessionId=0,  sessionPasswd是：

 

**
 *使用SessionId和Sessionpwd创建的ZooKeeper会话示例
 * @author zhangwei_david
 * @version $Id: ZKConstructorWithIdAndPasswdDemo.java, v 0.1 2015年5月1日 下午5:55:21 zhangwei_david Exp $
 */
public class ZKConstructorWithIdAndPasswdDemo implements Watcher {

    private static CountDownLatch connectedSemphore = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
        //创建Zookeeper会话实例
        ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000,
            new ZKConstructorWithIdAndPasswdDemo());
        // 输出当前会话的状态
        System.out.println("未使用id和密钥创建zk客户端的状态是：" + zookeeper.getState());
        long sessionId = zookeeper.getSessionId();
        byte[] sessionPasswd = zookeeper.getSessionPasswd();
        System.out.println("zk 客户端的sessionId=" + zookeeper.getSessionId() + ",  sessionPasswd是："
                + new String(zookeeper.getSessionPasswd()));
        zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new ZKConstructorWithIdAndPasswdDemo(),
            sessionId, sessionPasswd);
        System.out.println("使用id和密钥创建zk客户端的状态是：" + zookeeper.getState());

        zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new ZKConstructorWithIdAndPasswdDemo(),
            1L, "test".getBytes());
        System.out.println("使用错误id和密钥创建zk客户端的状态是：" + zookeeper.getState());
        try {
            // 当前闭锁在为0之前一直等待，除非线程中断
            connectedSemphore.await();
        } catch (Exception e) {
            System.out.println("Zookeeper session established");
        }
    }

    /**
     * @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.WatchedEvent)
     */
    public void process(WatchedEvent event) {
        System.out.println("Receive watched event:" + event);
        //如果客户端已经处于连接状态闭锁减去1
        if (KeeperState.SyncConnected == event.getState()) {
            connectedSemphore.countDown();
        }
    }
}

 

结果：

2015-05-01 19:14:34  [ main:0 ] - [ INFO ]  Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=5000 watcher=com.cathy.demo.zk.ZKConstructorWithIdAndPasswdDemo@1e717c2
未使用id和密钥创建zk客户端的状态是：CONNECTING
zk 客户端的sessionId=0,  sessionPasswd是：
2015-05-01 19:14:34  [ main:31 ] - [ INFO ]  Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=5000 watcher=com.cathy.demo.zk.ZKConstructorWithIdAndPasswdDemo@aa7bc2 sessionId=0 sessionPasswd=<hidden>
使用id和密钥创建zk客户端的状态是：CONNECTING
2015-05-01 19:14:34  [ main:31 ] - [ INFO ]  Initiating client connection, connectString=127.0.0.1:2181 sessionTimeout=5000 watcher=com.cathy.demo.zk.ZKConstructorWithIdAndPasswdDemo@ccd017 sessionId=1 sessionPasswd=<hidden>
使用错误id和密钥创建zk客户端的状态是：CONNECTING
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):31 ] - [ INFO ]  Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):31 ] - [ INFO ]  Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):31 ] - [ INFO ]  Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):47 ] - [ INFO ]  Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):47 ] - [ INFO ]  Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):47 ] - [ INFO ]  Socket connection established to 127.0.0.1/127.0.0.1:2181, initiating session
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):47 ] - [ INFO ]  Session establishment complete on server 127.0.0.1/127.0.0.1:2181, sessionid = 0x14d0ee972c70006, negotiated timeout = 5000
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):47 ] - [ INFO ]  Unable to reconnect to ZooKeeper service, session 0x1 has expired, closing socket connection
Receive watched event:WatchedEvent state:SyncConnected type:None path:null
Receive watched event:WatchedEvent state:Expired type:None path:null
2015-05-01 19:14:34  [ main-SendThread(127.0.0.1:2181):47 ] - [ INFO ]  Session establishment complete on server 127.0.0.1/127.0.0.1:2181, sessionid = 0x14d0ee972c70007, negotiated timeout = 5000
2015-05-01 19:14:34  [ main-EventThread:47 ] - [ INFO ]  EventThread shut down