Zk中的通知(Watcher)

ZooKeeper中實現對接點的監控,需要實現Watcher介面類,實現其中的process方法

public class WatcherDemo implements Watcher{
    public void process(WatchedEvent event) {

    }
}

WatcherEvent對如下狀態進行監聽
1. ZooKeeper會話狀態:Disconnected,SyncConnected，AuthFailed，ConnectedReadOnly,SaslAuthenticated和Expired
2. 事件型別:NodeCreated，NodeDeleted，NodeDataChanged,NodeChildrenChanged和None
3. 事件型別不是None時,返回一個znode路徑

如何設定一個監視點

對節點的建立設定監視點

/**
 * @author Eason
 * @create 2018-04-07 21:41
 **/
public class WatcherDemo implements Watcher {

    public void process(WatchedEvent event) {
        System.out.println("Enter the process method,the event is :"+event);
        Event.EventType type = event.getType();
        switch
 
 (type) {
            case NodeDeleted:
                System.out.println("新建節點:" + event.getPath());
        }
    }

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        String connectionString = "192.168.1.6:2181,192.168.1.6:2182,192.168.1.6:2183"
 
;
        ZooKeeper zooKeeper = new ZooKeeper(connectionString, 15 * 1000, new WatcherDemo(), false);
        zooKeeper.create("/myEphmeralPath1", "random".getBytes(), OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        TimeUnit.SECONDS.sleep(60);
        zooKeeper.close();
    }
}

執行完成後 控制檯輸出

Connected to the target VM, address: '127.0.0.1:50459', transport: 'socket'
Enter the process method,the event is :WatchedEvent state:SyncConnected type:None path:null
Disconnected from the target VM, address: '127.0.0.1:50459', transport: 'socket'

為什麼沒有監控到節點的建立呢?

首先,我們在建立ZK連線時指定了一個Watcher,但是這個Watcher指監控到了ZK連線成功的狀態SyncConnected.這裡同樣有一個問題,為什麼關閉客戶端時沒有監控到Disconnected事件呢.對於這個問題,個人理解是因為當前客戶端失去連線,Watcher失效,在debug模式下使用同一個Watcher可以監控到Disconnected事件

/**
 * @author Eason
 * @create 2018-04-07 21:41
 **/
public class WatcherDemo implements Watcher {

    public void process(WatchedEvent event) {
        System.out.println("Enter the process method,the event is :"+event);
        Event.EventType type = event.getType();
        switch (type) {
            case NodeDeleted:
                System.out.println("新建節點:" + event.getPath());
        }
    }

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        String connectionString = "192.168.1.6:2181,192.168.1.6:2182,192.168.1.6:2183";
        ZooKeeper zooKeeper = new ZooKeeper(connectionString, 15 * 1000, new WatcherDemo(), false);
        ZooKeeper zooKeeper1 = new ZooKeeper(connectionString, 15 * 1000, new WatcherDemo(), false);
        zooKeeper.create("/myEphmeralPath1", "random".getBytes(), OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        zooKeeper.close();
        TimeUnit.SECONDS.sleep(60);
    }
}

控制檯輸出

Enter the process method,the event is :WatchedEvent state:SyncConnected type:None path:null
Enter the process method,the event is :WatchedEvent state:SyncConnected type:None path:null
Enter the process method,the event is :WatchedEvent state:Disconnected type:None path:null
Enter the process method,the event is :WatchedEvent state:Expired type:None path:null

注意的是需要debug模式下才可以看到輸出結果,具體原因不清楚.

現在回到第一個問題,如何對建立節點設定監控呢？
首先看一下create有關的API

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

可以看到,create方法是沒有Watcher相關方法的,因此,基於判斷沒有就建立的思想,可以使用exist API來設定一個監視點

/**
 * @author Eason
 * @create 2018-04-07 21:41
 **/
public class WatcherDemo01 implements Watcher {

    public void process(WatchedEvent event) {
        System.out.println("Enter the process method,the event is :" + event);
        Event.EventType type = event.getType();
        switch (type) {
            case NodeCreated:
                System.out.println("新建節點:" + event.getPath());
            case NodeDeleted:
                System.out.println("刪除節點:" + event.getPath());
            case NodeDataChanged:
                System.out.println("修改節點:" + event.getPath());
            case NodeChildrenChanged:
                System.out.println("子節點:" + event);
        }
    }

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        WatcherDemo01 watcherDemo = new WatcherDemo01();
        String connectionString = "192.168.1.6:2181,192.168.1.6:2182,192.168.1.6:2183";
        ZooKeeper zooKeeper = new ZooKeeper(connectionString, 15 * 1000, watcherDemo, false);
        String path = "/watcherdemo01";
        String childPath = "/watcherdemo01/child01";

        Stat stat = zooKeeper.exists(path, watcherDemo);
        if (stat == null) {
            zooKeeper.create(path, "random".getBytes(), OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        if (stat != null) {
            zooKeeper.delete(path, stat.getVersion());
        }
        TimeUnit.SECONDS.sleep(5);
        zooKeeper.close();
    }
}

如何設定監控點

下面是對如何設定監控點的總結
- 對於NodeCreated 通過exists API設定
- 對於NodeDeleted 通過exists 和 getData()設定
- 對於NodeDataChanged 通過exists或getData設定
- 對於NodeChildrenChanged 通過getChildren設定

ZooKeeper回撥

首先要區分通知和回撥的區別,通知是ZooKeeper中註冊了監視點的客戶端收到的事件報告訊息,而回調是基於非同步思想的,通過回撥函式來確定操作的完成情況

建立節點的回撥通知

/**
 * @author Eason
 * @create 2018-04-07 22:47
 **/
public class CallBackDemo implements Watcher {
    public static void main(String[] args) throws InterruptedException, IOException {
        String connectionString = "192.168.1.6:2181,192.168.1.6:2182,192.168.1.6:2183";
        ZooKeeper zooKeeper = new ZooKeeper(connectionString, 15 * 1000, new CallBackDemo(), false);
        String path = "/callBackDemo";
        String params = "123";
        zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new AsyncCallback.StringCallback() {
            public void processResult(int rc, String path, Object ctx, String name) {
                System.out.println("Object ctx 引數對應於create 方法最後一個引數");
                System.out.println(ctx.toString());
                KeeperException.Code code = KeeperException.Code.get(rc);
                switch (code) {
                    case OK:
                    case NOAUTH:
                    case NONODE:
                    case APIERROR:
                    case NOTEMPTY:
                    case AUTHFAILED:
                    case BADVERSION:
                    case INVALIDACL:
                    case NODEEXISTS:
                    case NOTREADONLY:
                    case SYSTEMERROR:
                    case BADARGUMENTS:
                    case SESSIONMOVED:
                    case UNIMPLEMENTED:
                    case CONNECTIONLOSS:
                    case SESSIONEXPIRED:
                    case INVALIDCALLBACK:
                    case MARSHALLINGERROR:
                    case OPERATIONTIMEOUT:
                    case DATAINCONSISTENCY:
                    case RUNTIMEINCONSISTENCY:
                    case NOCHILDRENFOREPHEMERALS:
                }
            }
        }, params);
        TimeUnit.SECONDS.sleep(10);
    }

    public void process(WatchedEvent event) {

    }
}

注意Object ctx引數,具體狀態嗎不再贅述