public class WatcherTest implements Watcher {
	@Override
	public void process(WatchedEvent event) {
		// TODO Auto-generated method stub
		WatcherTest  w = new WatcherTest();
		ZooKeeper zk = new ZooKeeper(wx.getZkpath(),10000, w);	
	}
	
	public static void main(String[] args){
		WatcherTest  w = new WatcherTest();
		ZooKeeper	zk = new ZooKeeper(wx.getZkpath(), 10000, w);
	}
} 上面例子是把异常处理，逻辑处理等都省掉。watcher的应用很简单，主要有两步：继承 **Watcher** 接口，重写 **process** 回调函数。

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, byte[] sessionPasswd)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long sessionId, byte[] sessionPasswd, boolean canBeReadOnly)

/**
 * This Request processor actually applies any transaction associated with a
 * request and services any queries. It is always at the end of a
 * RequestProcessor chain (hence the name), so it does not have a nextProcessor
 * member.
 *
 * This RequestProcessor counts on ZooKeeperServer to populate the
 * outstandingRequests member of ZooKeeperServer.
 */
public class FinalRequestProcessor implements RequestProcessor 由源码注释得知，**FinalRequestProcessor**类实际是任何事务请求和任何查询的的最终处理类。也就是我们客户端对节点的set/get/delete/create/exists等操作最终都会运行到这里。

case OpCode.exists: {
	lastOp = "EXIS";
	// TODO we need to figure out the security requirement for this!
	ExistsRequest existsRequest = new ExistsRequest();
	ByteBufferInputStream.byteBuffer2Record(request.request,
	existsRequest);
	String path = existsRequest.getPath();
	if (path.indexOf('\0') != -1) {
	throw new KeeperException.BadArgumentsException();
	}
	Stat stat = zks.getZKDatabase().statNode(path, existsRequest
	.getWatch() ? cnxn : null);
	rsp = new ExistsResponse(stat);
	break;
} `existsRequest.getWatch() ? cnxn : null`此句是在调用exists API时，判断是否注册watcher，若是就返回 **cnxn**，**cnxn**是由此句代码`ServerCnxn cnxn = request.cnxn;`创建的。

/**
 * Interface to a Server connection - represents a connection from a client
 * to the server.
 */
public abstract class ServerCnxn implements Stats, Watcher 通过`ServerCnxn`类的源码注释得知，`ServerCnxn`是维持服务器与客户端的**tcp连接**与实现了 **watcher**。总的来说，ServerCnxn类创建的对象**cnxn**即包含了连接信息又包含watcher信息。

public abstract class ServerCnxn implements Stats, Watcher {
	.
	.
	public abstract void process(WatchedEvent event);
	Stat stat = zks.getZKDatabase().statNode(path, existsRequest.getWatch() ? cnxn : null); 
	//getZKDatabase实际上是获取是在zookeeper运行时的数据库。请看下面
	.
	.
}

/**
 * This class maintains the in memory database of zookeeper
 * server states that includes the sessions, datatree and the
 * committed logs. It is booted up  after reading the logs
 * and snapshots from the disk.
 */
public class ZKDatabase 通过源码注释得知**ZKDatabase**是在zookeeper运行时的数据库，在`FinalRequestProcessor`的case exists中会把existsRequest(exists请求传递给ZKDatabase)。

/**
 * the datatree for this zkdatabase
 * @return the datatree for this zkdatabase
 */
public DataTree getDataTree() {
return this.dataTree;
} **ZKDatabase**里面有这关键的一个函数是从zookeeper运行时展开的节点数型结构中搜索到合适的节点返回。

public Stat setData(String path, byte data[], int version, long zxid,long time) throws KeeperException.NoNodeException {
    Stat s = new Stat();
    DataNode n = nodes.get(path);
    if (n == null) {
    	throw new KeeperException.NoNodeException();
    }
    byte lastdata[] = null;
    synchronized (n) {
	    lastdata = n.data;
	    n.data = data;
	    n.stat.setMtime(time);
	    n.stat.setMzxid(zxid);
	    n.stat.setVersion(version);
	    n.copyStat(s);
    }
    // now update if the path is in a quota subtree.
    String lastPrefix = getMaxPrefixWithQuota(path);
    if(lastPrefix != null) {
      this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
      - (lastdata == null ? 0 : lastdata.length));
    }
    dataWatches.triggerWatch(path, EventType.NodeDataChanged); //触发事件
    return s;
}