Zookeeper源码分析：Watcher机制

1. 设置Watcher

使用Watcher需要先实现Watcher接口，并将实现类对象传递到指定方法中，如getChildren, exist等。Zookeeper允许在构造Zookeeper对象时候指定一个默认Watcher对象.getChildren和exit方法可以使用这个默认的Watcher对象，也可以指定一个新Watcher对象。

Code 1: Watcher接口

public interface Watcher {

/**
* Event的状态
*/
public interface Event {
/**
* 在事件发生时，ZooKeeper的状态
*/
public enum KeeperState {

@Deprecated
Unknown (-1),

Disconnected (0),

@Deprecated
NoSyncConnected (1),

SyncConnected (3),

AuthFailed (4),

ConnectedReadOnly (5),

SaslAuthenticated(6),

Expired (-112);

private final int intValue;

KeeperState( int intValue) {
this.intValue = intValue;
}

......
}

/**
* ZooKeeper中的事件
*/
public enum EventType {
None (-1),
NodeCreated (1),
NodeDeleted (2),
NodeDataChanged (3),
NodeChildrenChanged (4);

private final int intValue;     // Integer representation of value
// for sending over wire
EventType( int intValue) {
this.intValue = intValue;
}
......
}
}

//Watcher的回调方法
abstract public void process(WatchedEvent event);
}

Code 2: Zookeeper.getChildren（final String, Watcher)方法

public List<String> getChildren(final String path, Watcher watcher)
throws KeeperException, InterruptedException
{
final String clientPath = path;
PathUtils. validatePath(clientPath);

WatchRegistration wcb = null;
//如果watcher不等于null, 构建WatchRegistration对象，
//该对象描述了watcher和path之间的关系
if (watcher != null) {
wcb = new ChildWatchRegistration(watcher, clientPath);
}

//在传入的path加上root path前缀，构成服务器端的绝对路径
final String serverPath = prependChroot(clientPath);

//构建RequestHeader对象
RequestHeader h = new RequestHeader();
//设置操作类型为OpCode. getChildren
h.setType(ZooDefs.OpCode. getChildren);
//构建GetChildrenRequest对象
GetChildrenRequest request = new GetChildrenRequest();
//设置path
request.setPath(serverPath);
//设置是否使用watcher
request.setWatch(watcher != null);
//构建GetChildrenResponse对象
GetChildrenResponse response = new GetChildrenResponse();
//提交请求，并阻塞等待结果
ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
if (r.getErr() != 0) {
throw KeeperException.create(KeeperException.Code. get(r.getErr()),
clientPath);
}
return response.getChildren();
}

Follower的NIOServerCnxn类接到了Client的请求，会调用ZookeeperServer.processPacket()方法。该方法会构建一个Request对象，并调用第一个处理器FollowerRequestProcessor。

由于我们的请求只是一个读操作，而不是一个Quorum请求或者sync请求，所以FollowerRequestProcessor不需要调用Follower.request()方法将请求转给Leader，只需要将请求传递到下一个处理器CommitProcessor。

处理器CommitProcessor线程发现请求是读请求后，直接将Requet对象加入到toProcess队列中，在接下的循环中会调用FinalRequestProcessor.processRequest方法进行处理。

FinalRequestProcessor.processRequest方法最终会调用ZKDatabase中的读操作方法（如statNode和getData方法）， 而ZKDatabase的这些方法会最终调用DataTree类的方法来获取指定path的znode信息并返回给Client端，同时也会设置Watcher。

Code 3: FinalRequestProcessor对OpCode.getData请求的处理

case OpCode. getData: {
lastOp = "GETD";
GetDataRequest getDataRequest = new GetDataRequest();
ByteBufferInputStream. byteBuffer2Record(request.request,
getDataRequest);
//获得znode对象
DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath());
//n为null, 抛出NoNodeException异常
if (n == null) {
throw new KeeperException.NoNodeException();
}
Long aclL;
synchronized(n) {
aclL = n. acl;
}
//检查是否有读权限
PrepRequestProcessor. checkACL(zks, zks.getZKDatabase().convertLong(aclL),
ZooDefs.Perms. READ,
request. authInfo);
//构建状态对象stat
Stat stat = new Stat();
//获得指定path的znode数据，
//如果GetDataRequest.getWatcher()返回true, 将ServerCnxn类型对象cnxn传递进去。
//ServerCnxn是实现了Watcher接口
byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
getDataRequest. getWatch() ? cnxn : null);
//构建GetDataResponse对象
rsp = new GetDataResponse(b, stat);
break;
}

Code 4: DataTree.getData()方法

public byte[] getData(String path, Stat stat, Watcher watcher)
throws KeeperException.NoNodeException {
//从nodes map中获取指定path的DataNode对象
DataNode n = nodes.get(path);
//如果n为null, 则抛出NoNodeException异常
if (n == null) {
throw new KeeperException.NoNodeException();
}
synchronized (n) {
//将n的状态copy到stat中
n.copyStat(stat);
//如果watcher不会null, 则将(path, watcher)键值对放入dataWatchers Map里
if (watcher != null) {
dataWatches.addWatch(path, watcher);
}
//返回节点数据
return n.data ;
}
}

2. 修改znode数据触发Watcher

在Zookeeper二阶段提交的COMMIT阶段。当Follower从Leader那接收到一个写请求的Leader.COMMIT数据包，会调用FinalRequestProcessor.processRequest()方法。Leader本身在发送完Leader.COMMIT数据包，也会调用FinalRequestProcessor.processRequest()方法。

如果是setData修改数据请求，那么FinalRequestProcessor.processRequest()方法最终会调用到DataTree.setData方法将txn应用到指定znode上，同时触发Watcher，并发送notification给Client端。

其关SetData请求的时序图如下：





Code 5: DataTree.setData()方法

public Stat setData(String path, byte data[], int version, long zxid,
long time) throws KeeperException.NoNodeException {
Stat s = new Stat();
//根据path, 获得DataNode对象n
DataNode n = nodes.get(path);
//如果n为null, 则抛出NoNodeException异常
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 ));
}
//触发Watcher
dataWatches.triggerWatch(path, EventType.NodeDataChanged);
return s;
}

Code 6: WatchManage.triggerWatcher()方法，触发Watcher。

Set<Watcher> triggerWatch(String path, EventType type, Set<Watcher> supress) {
WatchedEvent e = new WatchedEvent(type,
KeeperState. SyncConnected, path);
HashSet<Watcher> watchers;
synchronized (this ) {
//从watchTable删除掉path对于的watcher
watchers = watchTable.remove(path);
if (watchers == null || watchers.isEmpty()) {
if (LOG .isTraceEnabled()) {
ZooTrace. logTraceMessage(LOG,
ZooTrace. EVENT_DELIVERY_TRACE_MASK,
"No watchers for " + path);
}
return null;
}
for (Watcher w : watchers) {
HashSet<String> paths = watch2Paths.get(w);
if (paths != null) {
paths.remove(path);
}
}
}
//循环处理所有关于path的Watcher, 这里Watcher对象实际上就是ServerCnxn类型对象
for (Watcher w : watchers) {
if (supress != null && supress.contains(w)) {
continue;
}
w.process(e);
}
return watchers;
}

Code 7: NIOServerCnxn.process方法，发送notification给Client端

synchronized public void process (WatchedEvent event) {
ReplyHeader h = new ReplyHeader(-1, -1L, 0);
if (LOG .isTraceEnabled()) {
ZooTrace. logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK ,
"Deliver event " + event + " to 0x"
+ Long. toHexString(this. sessionId)
+ " through " + this );
}

// Convert WatchedEvent to a type that can be sent over the wire
WatcherEvent e = event.getWrapper();

//发送notification给Client端
sendResponse(h, e, "notification");
}

3. 总结

Watcher具有one-time trigger的特性，在代码中我们也可以看到一个watcher被处理后会立即从watchTable中删掉。

本博客系博主原创，转载请附上原博客地址：http://blog.csdn.net/jeff_fangji/article/details/43910113