zookeeper watches

  Watches

所有的讀操作（getData(),getChildren(),exists()）都可以設(shè)置一個(gè)watch作為附加的操作

Watch的關(guān)鍵特性有

• 只觸發(fā)一次（one-timetrigger）

Watch當(dāng)數(shù)據(jù)變化時(shí)，只會被觸發(fā)一次，之后就會被移除，當(dāng)數(shù)據(jù)再次變化時(shí)，不會再觸發(fā)此watch

• 發(fā)送到客戶端（sentto the client）

通知客戶端是異步的，客戶端只會在它收到watch event時(shí)才會知道znode改變.通知客戶端的過程可能會發(fā)生異常，或者網(wǎng)絡(luò)延遲，但zookeeper會保證通知的一致性

• Watch為那些數(shù)據(jù)znode設(shè)置（the data for which the watch was set）

getData(),exists()設(shè)置數(shù)據(jù)監(jiān)控，getChildren()設(shè)置孩子監(jiān)控，

setData()會觸發(fā)數(shù)據(jù)監(jiān)控，成功create（）會觸發(fā)創(chuàng)建的znode數(shù)據(jù)監(jiān)控和父zonde的孩子監(jiān)控，成功的delete()zonde的數(shù)據(jù)監(jiān)控和孩子監(jiān)控、還有父zonde的孩子監(jiān)控

  Zookeeper 對watches一些保證

• Zookeeper客戶端保證所有的事情都是有序的（events,watches,異步回復(fù)）

• 客戶端先看到監(jiān)聽的事件，先于所監(jiān)聽的znode新的數(shù)據(jù)

• 監(jiān)聽事件的順序?qū)?yīng)zookeeper服務(wù)發(fā)生的更新時(shí)間順序

    關(guān)于監(jiān)聽器必須注意的

• 監(jiān)聽器只觸發(fā)一次

• 因?yàn)楸O(jiān)聽是一次性的，并且在獲取事件通知和發(fā)送新的請求設(shè)置監(jiān)聽器之間有延遲，所以不能完全相信每個(gè)事件的變化就是zonde當(dāng)前的狀態(tài)，有可能發(fā)生了多次變化之后你才收到監(jiān)聽事件的回復(fù)

• 同一個(gè)監(jiān)聽對象只會被一個(gè)通知事件觸發(fā)，比如同一個(gè)監(jiān)聽對象被注冊到一個(gè)exists和

GetDate，當(dāng)刪除這個(gè)節(jié)點(diǎn)的時(shí)候，監(jiān)聽對象只會被刪除事件觸發(fā)一次

• 當(dāng)和服務(wù)器斷開后，客戶端不會獲得任何監(jiān)聽器，直到重建連接，

以下是我根據(jù)官網(wǎng)例子簡化的實(shí)例，大家可以動手實(shí)踐以下，理解一下zookeeper的監(jiān)聽器的操作

package personal.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;

/**
 * comments()
 *
 * @author:czh ,mail:453775810@qq.com
 * @date 2015/11/4
 */
@Slf4j
public class Executor implementsWatcher,Runnable,AsyncCallback.StatCallback {
    Integer mutex=0;
    String znode;

    ZooKeeper zk;

    String filename;

    String exec[];


    public Executor(StringhostPort, String znode) throws KeeperException, IOException,InterruptedException {
        this.znode=znode;
        zk = new ZooKeeper(hostPort, 3000, this);
        zk.exists(znode,true, this, null);
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        String hostPort ="192.168.83.3:2181";
//       String hostPort="172.26.7.23:2181";
        String znode = "/dubbo";
        try {
            new Executor(hostPort,znode).run();
        } catch (Exceptione) {
            e.printStackTrace();
        }
    }

    /***************************************************************************
     * We do process any eventsourselves, we just need to forward them on.
     *
//     * @seeorg.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)
     */
    public void process(WatchedEventevent) {
        log.info("event.getPath():" + event.getPath() + "event.getState():" + event.getState() + "event.getType():" + event.getType());
    }

    public void run(){
        try {
            synchronized (mutex) {
                while (true) {
                    mutex.wait();
                }
            }
        } catch (InterruptedExceptione) {
            e.printStackTrace();
        }
    }

    /**
     * @param rc
     * @param path
     * @param ctx
     * @param stat
     */
    @Override
    public void proce***esult(int rc, String path, Object ctx, Stat stat) {
        log.info("proce***esult===========begin");
        boolean exists;
        log.info(String.valueOf(rc));
        switch (rc) {
            case KeeperException.Code.Ok:
                log.info("監(jiān)聽器設(shè)置成功");
                break;
            case KeeperException.Code.NoNode:
                log.info("znode["+znode+"]不存在");
                break;
            case KeeperException.Code.SessionExpired:
            case KeeperException.Code.NoAuth:
                return;
            default:
                // Retry errors
                zk.exists(znode, true, this, null);
                return;
        }
        log.info("proce***esult===========end");
    }
}