每一个Watcher具有如下属性：
1.KeeperState
2.EventType
3.path
4.process(WatchedEvent evnet)回掉方法
Watcher干嘛的？用户监听session的状态，数据节点的状态等。
watcher种类：defaultWatcher，非defaultWatcher
dafaultWatcher是在创建ZooKeeper对象时传递的watcher参数。非defaultWatcher只得是，在调用getData，getChildren，exists函数时传递的watcher对象。
二者的相同点在于：都可以监听数据节点的状态，比如在getData中使用了defaultWatcher，那么当监听的节点内容发生改变时，defaultWatcher就会收到通知。如果没有使用了非defaultWatcher，也是同样的。
而这的不同点在于：defaultWatcher会监听session的生命周期，比如session创建成功了，失效了等，而非defaultWatcher不具有这个职责。其次defaultWatcher并不与某一个节点路径相互关联。
notification event，非notification event
客户端需要接受服务器发送过来的消息，第一种消息是类似于Watcher回掉这种的，我们叫做notification，他的特点是服务器主动发送消息给客户端的，比如客户端a在数据节点a上设置了getData监听，当客户端b修改了节点a后，服务器主动发送NodeDataChanged消息给客户端a。第二中消息是类似于create，getData这种，他们向服务器发送对应的请求后，然后将请求放进到pendingQueue中，然后等待服务器的响应，当接受到服务器的响应后，再从pendingQueue中取出请求，然后进行回掉。对于第二中，只有两类请求是不需要服务器响应的，ping，autu。
Watcher和AsyncCallback区别
Watcher用于监听节点的，比如getData对数据节点a设置了watcher，那么当a的数据内容发生改变时，客户端会收到NodeDataChanged通知，然后进行watcher的回掉。
AsyncCallback是在以异步方式使用ZooKeeper APi时，用户获取api的处理结果的。而这具有本质的不同，不要混淆。
Watcher和AsyncCallback回调思路
为什么将Watcher和AsyncCallback放在一起说，这是因为从客户端的角度来看，他们都是异步的，客户端有个后台线程负责处理Watcher和AsyncCallback回调，这里我们好奇的是，客户端如何处理回调的。
一个最简单的方法就是生产者-消费者模式，生产者向push一个event到waitingEvents事件队列中，消费者线程从waitingEvents中取出event执行回调。这里的event可以是Watcher也可以是AsyncCallback。现在有几个问题需要解决：
1.生产者线程什么时候push event：对于Watcher回调来说，如果客户端在调用getData API时向节点A注册了一个Watcher，那么当节点A的数据发生改变时服务器会主动向客户端发出响应，客户端收到这个响应后，反序列化出WatchedEvent，然后找到所有在节点A注册的Watcher，将他们打包成一个event并push到waitingEvents。
对于AsyncCallback来说，如果客户端在调用getData API来获取节点A的数据并且传递了AsyncCallback，那么会将AsyncCallback保存到客户端的pendingQueue并向服务器获取节点A的数据内容，服务器在获取完节点A的数据内容后会向服务器发送响应，客户端拿到这个响应后取出pendingQueue中的AsyncCallback，然后打包成event 并push到waitingEvents。
Watcher和AsyncCallback回调实现
1.接受服务器的响应，并反序列化出ReplyHeader： 有一个单独的线程SendThread，负责接收服务器端的响应。假设他接受到的服务器传递过来的字节流失incomingBuffer。那么他做的第一步就应该将这个incomingBuffer反序列化出ReplyHeader。
2.判断响应类型：判断ReplyHeader是Wacher响应还是AsyncCallback响应：ReplyHeader.getXid()存储了响应类型。
2.1 如果是Wacher类型响应：从ReplyHeader中创建WatchedEvent，WatchedEvent里面存储了节点的路径，然后去WatcherManager中找到和这个节点相关联的所有Wacher，将他们写入到EventWaiting的waitingEvents中。
2.2 如果是AsyncCallback类型响应：从ReplyHeader中读取response，这个response描述了是Exists，setData，getData，getChildren，create.....中的哪一个异步回调。从pendingQueue中拿到Package，Package中的cb存储了AsyncCallback，也就是异步API的结果回调。最后将Package写入到EventThreadwaitingEvents中。
3.回调：EventThread从waitingEvents中取出
同步getData源码走读：
/**
* Return the data and the stat of the node of the given path.
* <p>
* If the watch is non-null and the call is successful (no exception is
* thrown), a watch will be left on the node with the given path. The watch
* will be triggered by a successful operation that sets data on the node, or
* deletes the node.
* <p>
* A KeeperException with error code KeeperException.NoNode will be thrown
* if no node with the given path exists.
*
* @param path the given path
* @param watcher explicit watcher
* @param stat the stat of the node
* @return the data of the node
* @throws KeeperException If the server signals an error with a non-zero error code
* @throws InterruptedException If the server transaction is interrupted.
* @throws IllegalArgumentException if an invalid path is specified
*/
public byte[] getData(final String path, Watcher watcher, Stat stat)
throws KeeperException, InterruptedException
{
final String clientPath = path;
PathUtils.validatePath(clientPath); // the watch contains the un-chroot path
WatchRegistration wcb = null;
if (watcher != null) {
wcb = new DataWatchRegistration(watcher, clientPath);
} final String serverPath = prependChroot(clientPath); RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.getData);
GetDataRequest request = new GetDataRequest();
request.setPath(serverPath);
request.setWatch(watcher != null);
GetDataResponse response = new GetDataResponse();
//提交请求，等待请求处理完毕
ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
if (r.getErr() != 0) {
throw KeeperException.create(KeeperException.Code.get(r.getErr()),
clientPath);
}
if (stat != null) {
DataTree.copyStat(response.getStat(), stat);
}
return response.getData();
}
public ReplyHeader submitRequest(RequestHeader h, Record request,
Record response, WatchRegistration watchRegistration)
throws InterruptedException {
ReplyHeader r = new ReplyHeader();
//提交请求到outgoingQueue，outgoingQueue里面的请求是将要发送到服务器的请求，ClientCnxnSocket的doIO()会从outgoingQueue中取出队列发送到服务器
Packet packet = queuePacket(h, r, request, response, null, null, null,
null, watchRegistration);
//阻塞等待，直到这个请求处理完成
synchronized (packet) {
while (!packet.finished) {
packet.wait();
}
}
return r;
}
1.将请求头，请求体，响应等送入队列，返回一个packet，然后等待packet完成。
Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
Record response, AsyncCallback cb, String clientPath,
String serverPath, Object ctx, WatchRegistration watchRegistration)
{
Packet packet = null; // Note that we do not generate the Xid for the packet yet. It is
// generated later at send-time, by an implementation of ClientCnxnSocket::doIO(),
// where the packet is actually sent.
//这里不会为这个packet生成xid，当ClientCnxnSocket::doIO从outgoingQueue中取出packet才生成xid
synchronized (outgoingQueue) {
packet = new Packet(h, r, request, response, watchRegistration);
packet.cb = cb;
packet.ctx = ctx;
packet.clientPath = clientPath;
packet.serverPath = serverPath;
//如果会话已经关闭
if (!state.isAlive() || closing) {
conLossPacket(packet);
} else {
// If the client is asking to close the session then
// mark as closing
//如果客户端请求关闭会话
if (h.getType() == OpCode.closeSession) {
closing = true;
}
outgoingQueue.add(packet);
}
}
sendThread.getClientCnxnSocket().wakeupCnxn();
return packet;
}
由于outgoingQueue并不是阻塞队列并且需要向其添加packet对象，所以需要对其synchronized。最后调用sendThread.getClientCnxnSocket().wakeupCnxn.
/**
* @return true if a packet was received
* @throws InterruptedException
* @throws IOException
*/
void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn)
throws InterruptedException, IOException {
SocketChannel sock = (SocketChannel) sockKey.channel();
if (sock == null) {
throw new IOException("Socket is null!");
}
//优先处理读响应
if (sockKey.isReadable()) {
int rc = sock.read(incomingBuffer);
if (rc < 0) {
throw new EndOfStreamException(
"Unable to read additional data from server sessionid 0x"
+ Long.toHexString(sessionId)
+ ", likely server has closed socket");
}
if (!incomingBuffer.hasRemaining()) {
incomingBuffer.flip();
if (incomingBuffer == lenBuffer) {
recvCount++;
readLength();
} else if (!initialized) {
readConnectResult();
enableRead();
if (findSendablePacket(outgoingQueue,
cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {
// Since SASL authentication has completed (if client is configured to do so),
// outgoing packets waiting in the outgoingQueue can now be sent.
enableWrite();
}
lenBuffer.clear();
incomingBuffer = lenBuffer;
updateLastHeard();
initialized = true;
} else {
sendThread.readResponse(incomingBuffer);
lenBuffer.clear();
incomingBuffer = lenBuffer;
updateLastHeard();
}
}
}
if (sockKey.isWritable()) {
synchronized(outgoingQueue) {
Packet p = findSendablePacket(outgoingQueue,
cnxn.sendThread.clientTunneledAuthenticationInProgress()); if (p != null) {
updateLastSend();
// If we already started writing p, p.bb will already exist
if (p.bb == null) {
if ((p.requestHeader != null) &&
(p.requestHeader.getType() != OpCode.ping) &&
(p.requestHeader.getType() != OpCode.auth)) {
p.requestHeader.setXid(cnxn.getXid());
}
p.createBB();
}
sock.write(p.bb);
if (!p.bb.hasRemaining()) {
sentCount++;
outgoingQueue.removeFirstOccurrence(p);
if (p.requestHeader != null
&& p.requestHeader.getType() != OpCode.ping
&& p.requestHeader.getType() != OpCode.auth) {
synchronized (pendingQueue) {
pendingQueue.add(p);
}
}
}
}
if (outgoingQueue.isEmpty()) {
// No more packets to send: turn off write interest flag.
// Will be turned on later by a later call to enableWrite(),
// from within ZooKeeperSaslClient (if client is configured
// to attempt SASL authentication), or in either doIO() or
// in doTransport() if not.
disableWrite();
} else {
// Just in case
enableWrite();
}
}
}
}
锁定outgoingQueue，获取一个可以发送的packet，如果则个packet还没有被发送，则p.bb！=null，然后生成xid。
xid初始值为1，每次发送一个packet时，都会递增xid，将其设置到请求头中。
最后，将packet写入到pendingQueue中。
void readResponse(ByteBuffer incomingBuffer) throws IOException {
ByteBufferInputStream bbis = new ByteBufferInputStream(
incomingBuffer);
BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
ReplyHeader replyHdr = new ReplyHeader();
replyHdr.deserialize(bbia, "header");
//虽然客户端会发送ping或者auth消息给服务器，但是客户端并不需要等待服务器的响应，也就是说他并没有将请求写入到pendingQueue中
if (replyHdr.getXid() == -2) {
// -2 is the xid for pings
if (LOG.isDebugEnabled()) {
LOG.debug("Got ping response for sessionid: 0x"
+ Long.toHexString(sessionId)
+ " after "
+ ((System.nanoTime() - lastPingSentNs) / 1000000)
+ "ms");
}
return;
}
if (replyHdr.getXid() == -4) {
// -4 is the xid for AuthPacket
if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
state = States.AUTH_FAILED;
eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None,
Watcher.Event.KeeperState.AuthFailed, null) );
}
if (LOG.isDebugEnabled()) {
LOG.debug("Got auth sessionid:0x"
+ Long.toHexString(sessionId));
}
return;
}
//notification类型的通知，watcher回掉相关逻辑。
if (replyHdr.getXid() == -1) {
// -1 means notification
if (LOG.isDebugEnabled()) {
LOG.debug("Got notification sessionid:0x"
+ Long.toHexString(sessionId));
}
WatcherEvent event = new WatcherEvent();
event.deserialize(bbia, "response");
// convert from a server path to a client path
if (chrootPath != null) {
String serverPath = event.getPath();
if(serverPath.compareTo(chrootPath)==0)
event.setPath("/");
else if (serverPath.length() > chrootPath.length())
event.setPath(serverPath.substring(chrootPath.length()));
else {
LOG.warn("Got server path " + event.getPath()
+ " which is too short for chroot path "
+ chrootPath);
}
}
WatchedEvent we = new WatchedEvent(event);
if (LOG.isDebugEnabled()) {
LOG.debug("Got " + we + " for sessionid 0x"
+ Long.toHexString(sessionId));
}
//根据<clientPath,EventType>从ZKWatcherManager中取出相关的wacher，然后封装成：
//<EventType(NodeDataChanged),Path(/abc),watche1 for getData,WatchedEvent>
//<EventType(NodeDataChanged),Path(/abc),watche1 for exists,WatchedEvent>
eventThread.queueEvent( we );
return;
} // If SASL authentication is currently in progress, construct and
// send a response packet immediately, rather than queuing a
// response as with other packets.
if (clientTunneledAuthenticationInProgress()) {
GetSASLRequest request = new GetSASLRequest();
request.deserialize(bbia,"token");
zooKeeperSaslClient.respondToServer(request.getToken(),
ClientCnxn.this);
return;
} Packet packet;
synchronized (pendingQueue) {
if (pendingQueue.size() == 0) {
throw new IOException("Nothing in the queue, but got "
+ replyHdr.getXid());
}
packet = pendingQueue.remove();
}
/*
* Since requests are processed in order, we better get a response
* to the first request!
*/
try {
if (packet.requestHeader.getXid() != replyHdr.getXid()) {
packet.replyHeader.setErr(
KeeperException.Code.CONNECTIONLOSS.intValue());
throw new IOException("Xid out of order. Got Xid "
+ replyHdr.getXid() + " with err " +
+ replyHdr.getErr() +
" expected Xid "
+ packet.requestHeader.getXid()
+ " for a packet with details: "
+ packet );
} packet.replyHeader.setXid(replyHdr.getXid());
packet.replyHeader.setErr(replyHdr.getErr());
packet.replyHeader.setZxid(replyHdr.getZxid());
if (replyHdr.getZxid() > 0) {
lastZxid = replyHdr.getZxid();
}
if (packet.response != null && replyHdr.getErr() == 0) {
packet.response.deserialize(bbia, "response");
} if (LOG.isDebugEnabled()) {
LOG.debug("Reading reply sessionid:0x"
+ Long.toHexString(sessionId) + ", packet:: " + packet);
}
} finally {
//客户端发送的非ping,auth请求，比如getData，setData等，注意事项，如果客户端注册了新的wather，需要将这个watcher保存到ZKWatcherManager中。如果是同步调用，直接通知客户端完成即可，如果是异步调用，就需要进行AsyncCallback回掉。
finishPacket(packet);
}
}
private void finishPacket(Packet p) {
//如果用户显示的注册了watcher，比如在getData中注册了非默认watcher，那么就将watcher添加到ZKWatcherManager中去，很重要的一步哦
//如果没有这一步，也就没有后面的watcher回掉了。
if (p.watchRegistration != null) {
p.watchRegistration.register(p.replyHeader.getErr());
}
//cb就是AsnycCallback，如果为null，表明是同步调用的接口，不需要异步回掉，因此，直接notifyAll即可。
if (p.cb == null) {
synchronized (p) {
p.finished = true;
p.notifyAll();
}
} else {
p.finished = true;
eventThread.queuePacket(p);
}
}
如果watchRegistration！=null，
如果p.cb==null,说明这个请求已经处理完毕了，因此通知packet完成即可。
最后，将packet放进waitingEvents中，然后EventThread就可以从waitingEvents中取出packets，然后执行客户端逻辑。
private void processEvent(Object event) {
try {
//watcher回掉逻辑
if (event instanceof WatcherSetEventPair) {
// each watcher will process the event
WatcherSetEventPair pair = (WatcherSetEventPair) event;
for (Watcher watcher : pair.watchers) {
try {
watcher.process(pair.event);
} catch (Throwable t) {
LOG.error("Error while calling watcher ", t);
}
}
} else {
//create，getData等需要异步回掉的
Packet p = (Packet) event;
int rc = 0;
String clientPath = p.clientPath;
if (p.replyHeader.getErr() != 0) {
rc = p.replyHeader.getErr();
}
if (p.cb == null) {
LOG.warn("Somehow a null cb got to EventThread!");
} else if (p.response instanceof ExistsResponse
|| p.response instanceof SetDataResponse
|| p.response instanceof SetACLResponse) {
StatCallback cb = (StatCallback) p.cb;
if (rc == 0) {
if (p.response instanceof ExistsResponse) {
cb.processResult(rc, clientPath, p.ctx,
((ExistsResponse) p.response)
.getStat());
} else if (p.response instanceof SetDataResponse) {
cb.processResult(rc, clientPath, p.ctx,
((SetDataResponse) p.response)
.getStat());
} else if (p.response instanceof SetACLResponse) {
cb.processResult(rc, clientPath, p.ctx,
((SetACLResponse) p.response)
.getStat());
}
} else {
cb.processResult(rc, clientPath, p.ctx, null);
}
} else if (p.response instanceof GetDataResponse) {
DataCallback cb = (DataCallback) p.cb;
GetDataResponse rsp = (GetDataResponse) p.response;
if (rc == 0) {
cb.processResult(rc, clientPath, p.ctx, rsp
.getData(), rsp.getStat());
} else {
cb.processResult(rc, clientPath, p.ctx, null,
null);
}
} else if (p.response instanceof GetACLResponse) {
ACLCallback cb = (ACLCallback) p.cb;
GetACLResponse rsp = (GetACLResponse) p.response;
if (rc == 0) {
cb.processResult(rc, clientPath, p.ctx, rsp
.getAcl(), rsp.getStat());
} else {
cb.processResult(rc, clientPath, p.ctx, null,
null);
}
} else if (p.response instanceof GetChildrenResponse) {
ChildrenCallback cb = (ChildrenCallback) p.cb;
GetChildrenResponse rsp = (GetChildrenResponse) p.response;
if (rc == 0) {
cb.processResult(rc, clientPath, p.ctx, rsp
.getChildren());
} else {
cb.processResult(rc, clientPath, p.ctx, null);
}
} else if (p.response instanceof GetChildren2Response) {
Children2Callback cb = (Children2Callback) p.cb;
GetChildren2Response rsp = (GetChildren2Response) p.response;
if (rc == 0) {
cb.processResult(rc, clientPath, p.ctx, rsp
.getChildren(), rsp.getStat());
} else {
cb.processResult(rc, clientPath, p.ctx, null, null);
}
} else if (p.response instanceof CreateResponse) {
StringCallback cb = (StringCallback) p.cb;
CreateResponse rsp = (CreateResponse) p.response;
if (rc == 0) {
cb.processResult(rc, clientPath, p.ctx,
(chrootPath == null
? rsp.getPath()
: rsp.getPath()
.substring(chrootPath.length())));
} else {
cb.processResult(rc, clientPath, p.ctx, null);
}
} else if (p.cb instanceof VoidCallback) {
VoidCallback cb = (VoidCallback) p.cb;
cb.processResult(rc, clientPath, p.ctx);
}
}
} catch (Throwable t) {
LOG.error("Caught unexpected throwable", t);
}
}
}
客户端在节点a注册了如下watcher：
getData(a,watcher1)
exists(a,watcher2)
getChildren(a,watcher3)
因此与a关联的watcher有watcher1和watcher2。如果节点a被修改了，那么客户端会收到notification 类型的通知，这里应是NodeDataChanged事件类型，此时，客户端需要回掉watcher1和watcher2.也就是说，他需要根据<EventType,clientPath>来找到与节点a对应的所有watcher。
作者：FrancisWang
邮箱：franciswbs@163.com
出处：http://www.cnblogs.com/francisYoung/
本文地址：http://www.cnblogs.com/francisYoung/p/5225703.html
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。