深入浅出Zookeeper之三Exists请求和处理

前一篇介绍了zookeeper的client和server之间session是如何建立的。在DataMonitor的例子中，DataMonitor通过exists异步接口和server端交互，本文将介绍exists操作是如何完成。

dataMonitor开始exist操作

 

   public void exists(final String path, Watcher watcher,
           StatCallback cb, Object ctx)
   {
        ......
//exist请求头
       RequestHeader h = new RequestHeader();
       h.setType(ZooDefs.OpCode.exists);
//请求体
       ExistsRequest request = new ExistsRequest();
       request.setPath(serverPath);
       request.setWatch(watcher != null);
       SetDataResponse response = new SetDataResponse();
//添加到发送队列
       cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
               clientPath, serverPath, ctx, wcb);
   }

 

 添加过程

 

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.
        synchronized (outgoingQueue) {
        //转换成Packet
            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);
            }
        }
    //唤醒下Selector，快点处理
        sendThread.getClientCnxnSocket().wakeupCnxn();
        return packet;
    }

 

 接下来还是回到SendThread的发送过程，之前Session建立时已经分析过，这里有一点要注意下：

 

//重要的业务请求，需要设置事务id
 if ((p.requestHeader != null) &&
                                (p.requestHeader.getType() != OpCode.ping) &&
                                (p.requestHeader.getType() != OpCode.auth)) {
                            p.requestHeader.setXid(cnxn.getXid());
                        }

 接下来Server端IO线程拿到请求，处理，过程和之前session建立时一样，就不赘述了。变化的是后续的处理链。

PrepRequestProcessor预处理

//All the rest don't need to create a Txn - just verify session
/读请求，不需要创建事务，只是检查了下session是否还在，此时事务头和事务体都是null
           case OpCode.sync:
           case OpCode.exists:
           case OpCode.getData:
           case OpCode.getACL:
           case OpCode.getChildren:
           case OpCode.getChildren2:
           case OpCode.ping:
           case OpCode.setWatches:
               zks.sessionTracker.checkSession(request.sessionId,
                       request.getOwner());
               break;

 SyncRequestProcessor处理逻辑之前已经分析过了，这里就挑重点说一下

//试图将其写log，由于ExistsRequest并不是一个事务型请求，所以这里直接返回false，也就是说ExistsRequest不会被记录到log文件中
zks.getZKDatabase().append(si)

 接下来FinalRequestProcessor处理，由于不是事务型请求，省了很多步骤，直接进入switch处理：

 

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();
                }
        //拿对应node的状态，并设置是否watch
                Stat stat = zks.getZKDatabase().statNode(path, existsRequest
                        .getWatch() ? cnxn : null);
        //结果
                rsp = new ExistsResponse(stat);
                break;
            }
......
//当前处理zxid
 long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
//构造相应头
        ReplyHeader hdr =
            new ReplyHeader(request.cxid, lastZxid, err.intValue());

        zks.serverStats().updateLatency(request.createTime);
        cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp,
                    request.createTime, System.currentTimeMillis());

        try {
    //写回响应
            cnxn.sendResponse(hdr, rsp, "response");

 

 statNode过程

 

//此处的watcher就是对应ServerCnxn，代表是哪个client，server端需要notify的时候，直接往对应ServerCnxn写数据即可
public Stat statNode(String path, Watcher watcher)
            throws KeeperException.NoNodeException {
        Stat stat = new Stat();
        DataNode n = nodes.get(path);
        if (watcher != null) {
        //对于exists请求，需要监听data变化事件，添加watcher
            dataWatches.addWatch(path, watcher);
        }
        if (n == null) {
            throw new KeeperException.NoNodeException();
        }
        synchronized (n) {
            n.copyStat(stat);
            return stat;
        }
    }

 

好了，以上server端就完成了ExistsRequest的处理了。接下来client端SendThread收到ExistsResponse进行处理

      if (sockKey.isReadable()) {
          int rc = sock.read(incomingBuffer);
......
 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;
              }
//此时client session已经建立，init完成
else {
                  sendThread.readResponse(incomingBuffer);
                  lenBuffer.clear();
                  incomingBuffer = lenBuffer;
                  updateLastHeard();
              }
          }
      }

  具体读取：

 

void readResponse(ByteBuffer incomingBuffer) throws IOException {
            ByteBufferInputStream bbis = new ByteBufferInputStream(
                    incomingBuffer);
            BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
            ReplyHeader replyHdr = new ReplyHeader();
        //先读响应头，先对特殊的xid进行处理
            replyHdr.deserialize(bbia, "header");
            ......
        //从头拿之前入队的ExistsRequest请求Packet，由于client和server都是单线程处理，多队列处理，所以认为全局有序
            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());
        //更新client端的最新zxid
                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 {
        //处理packet，主要是注册watcher，回调，触发事件
                finishPacket(packet);
            }
        }

  finishPacket过程

   private void finishPacket(Packet p) {
//注册watcher，如果是exists请求，则注册到dataWatches中
       if (p.watchRegistration != null) {
           p.watchRegistration.register(p.replyHeader.getErr());
       }
//如果是同步接口，则唤醒等待的业务线程
       if (p.cb == null) {
           synchronized (p) {
               p.finished = true;
               p.notifyAll();
           }
       }
//如果是异步请求，则发送异步事件
else {
           p.finished = true;
           eventThread.queuePacket(p);
       }
   }

 EventThread端

else if (p.response instanceof ExistsResponse
                          || p.response instanceof SetDataResponse
                          || p.response instanceof SetACLResponse) {
                      StatCallback cb = (StatCallback) p.cb;
            //如果成功，增加node stat的回调参数
                      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());
                          }
                      }
            //如果响应失败，stat为null
            else {
                          cb.processResult(rc, clientPath, p.ctx, null);
                      }
                  }

在DataMonitor例子中，它本身就是一个StatCallback

public void processResult(int rc, String path, Object ctx, Stat stat) {
        boolean exists;
        switch (rc) {
        case Code.Ok:
            exists = true;
            break;
        case Code.NoNode:
            exists = false;
            break;
        case Code.SessionExpired:
        case Code.NoAuth:
            dead = true;
            listener.closing(rc);
            return;
        default:
            // Retry errors
            zk.exists(znode, true, this, null);
            return;
        }

 Exists过程大致就是上面描述的，主要注意点：

1.客户端Request发送完之后会进入Pending队列，等待响应之后拿出来继续处理

2.同步接口是使用Packet.wait()实现的

3.server端exists操作不是事务型的操作，不会写入log

4.server端的watcher就是一个客户端连接ServerCxcn，代表一个客户端，notify的时候直接往连接里写数据即可