zookeeper watcher

zookeeper watcher

All of the read operations in ZooKeeper - getData(), getChildren(), and exists() - have the option of setting a watch as a side effect. Here is ZooKeeper's definition of a watch: a watch event is one-time trigger, sent to the client that set the watch, which occurs when the data for which the watch was set changes。

getData,getChildren(),exists()这三个方法能够针对参数中的path设置watcher，当path对应的Node 有相应变化时，server端会给对应的设置了watcher的client发送一个一次性的触发通知事件。客户端在收到这个触发通知事件后，能够根据本身的业务逻辑进行相应地处理。

注意这个watcher的功能是一次性的，若是还想继续获得watcher通知，在处理完事件后，要从新注册。

下面就作几个demo来体会一下。

首先看一下WatchedEvent，该事件有state、type和path属性，以下，

public class WatchedEvent {
    final private KeeperState keeperState;
    final private EventType eventType;
    private String path;
    ........
}

那么state表示什么含义，type又表示什么含义：KeeperState 的注解

public enum KeeperState {
    /** Unused, this state is never generated by the server */
    @Deprecated
    Unknown (-1),

    /** The client is in the disconnected state - it is not connected
     * to any server in the ensemble. */
    Disconnected (0),

    /** Unused, this state is never generated by the server */
    @Deprecated
    NoSyncConnected (1),

    /** The client is in the connected state - it is connected
     * to a server in the ensemble（全体，总效果） (one of the servers specified
     * in the host connection parameter during ZooKeeper client
     * creation). */
    SyncConnected (3),

    /**
     * Auth failed state
     */
    AuthFailed (4),

    /**
     * The client is connected to a read-only server, that is the
     * server which is not currently connected to the majority.
     * The only operations allowed after receiving this state is
     * read operations.
     * This state is generated for read-only clients only since
     * read/write clients aren't allowed to connect to r/o servers.
     */
    ConnectedReadOnly (5),

    /**
     * SaslAuthenticated: used to notify clients that they are SASL-authenticated,
     * so that they can perform Zookeeper actions with their SASL-authorized permissions.
     */
    SaslAuthenticated(6),

    /** The serving cluster has expired this session. The ZooKeeper
     * client connection (the session) is no longer valid. You must
     * create a new client connection (instantiate a new ZooKeeper
     * instance) if you with to access the ensemble. */
    Expired (-112);
    ........
}

再看EventType，

/**
 * Enumeration of types of events that may occur on the ZooKeeper
 */
public enum EventType {
    None(-1),
    NodeCreated(1),
    NodeDeleted(2),
    NodeDataChanged(3),
    NodeChildrenChanged(4);
}

ok，很好明白。下面就作一个示例，

package com.usfot;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;

/**
 * Created by liyanxin on 2015/3/17.
 */
public class ZookeeperWathcherDemo {

    //互斥锁
    private static Integer mutex = new Integer(-1);

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

        ZooKeeper zk = new ZooKeeper("127.0.0.1:2181", 300000, new Watcher() {
            // 监控全部被触发的事件
            public void process(WatchedEvent event) {
                System.out.println("状态:" + event.getState() + "|类型:" + event.getType() +
                        "|Wrapper:" + event.getWrapper() + "|Path:" + event.getPath());
            }
        });

        // 建立一个目录节点
        zk.create("/testRootPath", "testRootData".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
        // 建立一个子目录节点
        zk.create("/testRootPath/testChildPathOne", "testChildDataOne".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);


        zk.getData("/testRootPath/testChildPathOne", new Watcher() {
            @Override
            public void process(WatchedEvent event) {

                if (event.getState() == Event.KeeperState.SyncConnected
                        && event.getType() == Event.EventType.NodeDataChanged) {
                    System.out.println("znode:/testRootPath/testChildPathOne data change");
                    synchronized (mutex) {
                        mutex.notify();
                    }
                }
            }
        }, null);

        synchronized (mutex) {
            mutex.wait(); //阻塞直到notify
        }
        zk.close();
    }
}

这段代码对路径为/testRootPath/testChildPathOne的znode注册了一个watcher，该watcher监听的state为SyncConnected，而且类型为NodeDataChanged，当发生这个时间后，打印消息，而且notify条件变量mutex，使程序结束。

是如何使该znode发生改变呢，以下打开zk的客户端，

[zk: localhost:2181(CONNECTED) 26] set /testRootPath/testChildPathOne mydata
cZxid = 0x700000044
ctime = Tue Mar 17 17:46:12 CST 2015
mZxid = 0x700000045
mtime = Tue Mar 17 17:47:49 CST 2015
pZxid = 0x700000044
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
[zk: localhost:2181(CONNECTED) 27]

这时那段程序打印消息，以下，

状态:SyncConnected|类型:None|Wrapper:-1,3,
|Path:null
znode:/testRootPath/testChildPathOne data change

Process finished with exit code 0

参考：http://luzengyi.blog.163.com/blog/static/529188201064113744373/

==================END==================