Zookeeper-watcher机制源码分析（二）

服务端接收请求处理流程

在前一篇文章中Zookeeper-watcher机制源码分析（一）说过Watcher的基本流程，在此文中详细剖析服务端几首请求处理流程。

服务端有一个NettyServerCnxn类，用来处理客户端发送过来的请求 

NettyServerCnxn 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109

public void receiveMessage(ChannelBuffer message) {         try {             while(message.readable() && !throttled) {                 if (bb != null) { //ByteBuffer不为空                     if (LOG.isTraceEnabled()) {                         LOG.trace("message readable " + message.readableBytes()                                 + " bb len " + bb.remaining() + " " + bb);                         ByteBuffer dat = bb.duplicate();                         dat.flip();                         LOG.trace(Long.toHexString(sessionId)                                 + " bb 0x"                                 + ChannelBuffers.hexDump(                                         ChannelBuffers.copiedBuffer(dat)));                     }                     //bb剩余空间大于message中可读字节大小                     if (bb.remaining() > message.readableBytes()) {                         int newLimit = bb.position() + message.readableBytes();                         bb.limit(newLimit);                     }                     // 将message写入bb中                     message.readBytes(bb);                     bb.limit(bb.capacity());                       if (LOG.isTraceEnabled()) {                         LOG.trace("after readBytes message readable "                                 + message.readableBytes()                                 + " bb len " + bb.remaining() + " " + bb);                         ByteBuffer dat = bb.duplicate();                         dat.flip();                         LOG.trace("after readbytes "                                 + Long.toHexString(sessionId)                                 + " bb 0x"                                 + ChannelBuffers.hexDump(                                         ChannelBuffers.copiedBuffer(dat)));                     }                     if (bb.remaining() == 0) { // 已经读完message，表示内容已经所有接收                         packetReceived(); // 统计接收信息                         bb.flip();                           ZooKeeperServer zks = this.zkServer;                         if (zks == null || !zks.isRunning()) {//Zookeeper服务器为空 ,说明服务端挂了                             throw new IOException("ZK down");                         }                         if (initialized) {                             //处理客户端传过来的数据包                             zks.processPacket(this, bb);                               if (zks.shouldThrottle(outstandingCount.incrementAndGet())) {                                 disableRecvNoWait();                             }                         } else {                             LOG.debug("got conn req request from "                                     + getRemoteSocketAddress());                             zks.processConnectRequest(this, bb);                             initialized = true;                         }                         bb = null;                     }                 } else { //bb为null的状况，你们本身去看，我就不细讲了                     if (LOG.isTraceEnabled()) {                         LOG.trace("message readable "                                 + message.readableBytes()                                 + " bblenrem " + bbLen.remaining());                         ByteBuffer dat = bbLen.duplicate();                         dat.flip();                         LOG.trace(Long.toHexString(sessionId)                                 + " bbLen 0x"                                 + ChannelBuffers.hexDump(                                         ChannelBuffers.copiedBuffer(dat)));                     }                       if (message.readableBytes() < bbLen.remaining()) {                         bbLen.limit(bbLen.position() + message.readableBytes());                     }                     message.readBytes(bbLen);                     bbLen.limit(bbLen.capacity());                     if (bbLen.remaining() == 0) {                         bbLen.flip();                           if (LOG.isTraceEnabled()) {                             LOG.trace(Long.toHexString(sessionId)                                     + " bbLen 0x"                                     + ChannelBuffers.hexDump(                                             ChannelBuffers.copiedBuffer(bbLen)));                         }                         int len = bbLen.getInt();                         if (LOG.isTraceEnabled()) {                             LOG.trace(Long.toHexString(sessionId)                                     + " bbLen len is " + len);                         }                           bbLen.clear();                         if (!initialized) {                             if (checkFourLetterWord(channel, message, len)) {                                 return;                             }                         }                         if (len < 0 || len > BinaryInputArchive.maxBuffer) {                             throw new IOException("Len error " + len);                         }                         bb = ByteBuffer.allocate(len);                     }                 }             }         } catch(IOException e) {             LOG.warn("Closing connection to " + getRemoteSocketAddress(), e);             close();         }     }

ZookeeperServer-zks.processPacket(this, bb);

处理客户端传送过来的数据包

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70

public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {         // We have the request, now process and setup for next         InputStream bais = new ByteBufferInputStream(incomingBuffer);         BinaryInputArchive bia = BinaryInputArchive.getArchive(bais);         RequestHeader h = new RequestHeader();         h.deserialize(bia, "header"); //反序列化客户端header头信息         // Through the magic of byte buffers, txn will not be         // pointing         // to the start of the txn         incomingBuffer = incomingBuffer.slice();         if (h.getType() == OpCode.auth) { //判断当前操做类型，若是是auth操做，则执行下面的代码             LOG.info("got auth packet " + cnxn.getRemoteSocketAddress());             AuthPacket authPacket = new AuthPacket();             ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket);             String scheme = authPacket.getScheme();             ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(scheme);             Code authReturn = KeeperException.Code.AUTHFAILED;             if(ap != null) {                 try {                     authReturn = ap.handleAuthentication(new ServerAuthenticationProvider.ServerObjs(this, cnxn), authPacket.getAuth());                 } catch(RuntimeException e) {                     LOG.warn("Caught runtime exception from AuthenticationProvider: " + scheme + " due to " + e);                     authReturn = KeeperException.Code.AUTHFAILED;                 }             }             if (authReturn == KeeperException.Code.OK) {                 if (LOG.isDebugEnabled()) {                     LOG.debug("Authentication succeeded for scheme: " + scheme);                 }                 LOG.info("auth success " + cnxn.getRemoteSocketAddress());                 ReplyHeader rh = new ReplyHeader(h.getXid(), 0,                         KeeperException.Code.OK.intValue());                 cnxn.sendResponse(rh, null, null);             } else {                 if (ap == null) {                     LOG.warn("No authentication provider for scheme: "                             + scheme + " has "                             + ProviderRegistry.listProviders());                 } else {                     LOG.warn("Authentication failed for scheme: " + scheme);                 }                 // send a response...                 ReplyHeader rh = new ReplyHeader(h.getXid(), 0,                         KeeperException.Code.AUTHFAILED.intValue());                 cnxn.sendResponse(rh, null, null);                 // ... and close connection                 cnxn.sendBuffer(ServerCnxnFactory.closeConn);                 cnxn.disableRecv();             }             return;         } else { //若是不是受权操做，再判断是否为sasl操做             if (h.getType() == OpCode.sasl) {                 Record rsp = processSasl(incomingBuffer,cnxn);                 ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue());                 cnxn.sendResponse(rh,rsp, "response"); // not sure about 3rd arg..what is it?                 return;             }             else {//最终进入这个代码块进行处理                 //封装请求对象                 Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(),                   h.getType(), incomingBuffer, cnxn.getAuthInfo());                 si.setOwner(ServerCnxn.me);                 // Always treat packet from the client as a possible                 // local request.                 setLocalSessionFlag(si);                 submitRequest(si); //提交请求             }         }         cnxn.incrOutstandingRequests(h);     }

　　

submitRequest

负责在服务端提交当前请求

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39

public void submitRequest(Request si) {         if (firstProcessor == null) { //processor处理器，request过来之后会经历一系列处理器的处理过程             synchronized (this) {                 try {                     // Since all requests are passed to the request                     // processor it should wait for setting up the request                     // processor chain. The state will be updated to RUNNING                     // after the setup.                     while (state == State.INITIAL) {                         wait(1000);                     }                 } catch (InterruptedException e) {                     LOG.warn("Unexpected interruption", e);                 }                 if (firstProcessor == null || state != State.RUNNING) {                     throw new RuntimeException("Not started");                 }             }         }         try {             touch(si.cnxn);             boolean validpacket = Request.isValid(si.type); //判断是否合法             if (validpacket) {                 firstProcessor.processRequest(si);  调用firstProcessor发起请求，而这个firstProcess是一个接口，有多个实现类，具体的调用链是怎么样的？往下看吧                 if (si.cnxn != null) {                     incInProcess();                 }             } else {                 LOG.warn("Received packet at server of unknown type " + si.type);                 new UnimplementedRequestProcessor().processRequest(si);             }         } catch (MissingSessionException e) {             if (LOG.isDebugEnabled()) {                 LOG.debug("Dropping request: " + e.getMessage());             }         } catch (RequestProcessorException e) {             LOG.error("Unable to process request:" + e.getMessage(), e);         }     }

　　

firstProcessor的请求链组成

1.firstProcessor的初始化是在ZookeeperServer的setupRequestProcessor中完成的，代码以下

1 2 3 4 5 6 7

protected void setupRequestProcessors() {      RequestProcessor finalProcessor = new FinalRequestProcessor(this);      RequestProcessor syncProcessor = new SyncRequestProcessor(this, finalProcessor);      ((SyncRequestProcessor)syncProcessor).start();      firstProcessor = new PrepRequestProcessor(this, syncProcessor);//须要注意的是，PrepRequestProcessor中传递的是一个syncProcessor      ((PrepRequestProcessor)firstProcessor).start();  }

　　

从上面咱们能够看到firstProcessor的实例是一个PrepRequestProcessor，而这个构造方法中又传递了一个Processor构成了一个调用链。

RequestProcessor syncProcessor = new SyncRequestProcessor(this, finalProcessor);

而syncProcessor的构造方法传递的又是一个Processor，对应的是FinalRequestProcessor

2.因此整个调用链是PrepRequestProcessor -> SyncRequestProcessor ->FinalRequestProcessor

PredRequestProcessor.processRequest(si);

经过上面了解到调用链关系之后，咱们继续再看firstProcessor.processRequest(si)； 会调用到PrepRequestProcessor

 

1 2 3

public void processRequest(Request request) {     submittedRequests.add(request); }

　　

唉，很奇怪，processRequest只是把request添加到submittedRequests中，根据前面的经验，很天然的想到这里又是一个异步操做。而subittedRequests又是一个阻塞队列

LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();

而PrepRequestProcessor这个类又继承了线程类，所以咱们直接找到当前类中的run方法以下

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

public void run() {         try {             while (true) {                 Request request = submittedRequests.take(); //ok，从队列中拿到请求进行处理                 long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;                 if (request.type == OpCode.ping) {                     traceMask = ZooTrace.CLIENT_PING_TRACE_MASK;                 }                 if (LOG.isTraceEnabled()) {                     ZooTrace.logRequest(LOG, traceMask, 'P', request, "");                 }                 if (Request.requestOfDeath == request) {                     break;                 }                 pRequest(request); //调用pRequest进行预处理             }         } catch (RequestProcessorException e) {             if (e.getCause() instanceof XidRolloverException) {                 LOG.info(e.getCause().getMessage());             }             handleException(this.getName(), e);         } catch (Exception e) {             handleException(this.getName(), e);         }         LOG.info("PrepRequestProcessor exited loop!");     }

　　

pRequest

预处理这块的代码太长，就很差贴了。前面的N行代码都是根据当前的OP类型进行判断和作相应的处理，在这个方法中的最后一行中，咱们会看到以下代码

nextProcessor.processRequest(request);

1	nextProcessor.processRequest(request);

很显然，nextProcessor对应的应该是SyncRequestProcessor

SyncRequestProcessor. processRequest

1 2 3 4

public void processRequest(Request request) {     // request.addRQRec(">sync");     queuedRequests.add(request); }

这个方法的代码也是同样，基于异步化的操做，把请求添加到queuedRequets中，那么咱们继续在当前类找到run方法

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74

public void run() {         try {             int logCount = 0;               // we do this in an attempt to ensure that not all of the servers             // in the ensemble take a snapshot at the same time             int randRoll = r.nextInt(snapCount/2);             while (true) {                 Request si = null;                 //从阻塞队列中获取请求                 if (toFlush.isEmpty()) {                     si = queuedRequests.take();                  } else {                     si = queuedRequests.poll();                     if (si == null) {                         flush(toFlush);                         continue;                     }                 }                 if (si == requestOfDeath) {                     break;                 }                 if (si != null) {                     // track the number of records written to the log                     //下面这块代码，粗略看来是触发快照操做，启动一个处理快照的线程                     if (zks.getZKDatabase().append(si)) {                         logCount++;                         if (logCount > (snapCount / 2 + randRoll)) {                             randRoll = r.nextInt(snapCount/2);                             // roll the log                             zks.getZKDatabase().rollLog();                             // take a snapshot                             if (snapInProcess != null && snapInProcess.isAlive()) {                                 LOG.warn("Too busy to snap, skipping");                             } else {                                 snapInProcess = new ZooKeeperThread("Snapshot Thread") {                                         public void run() {                                             try {                                                 zks.takeSnapshot();                                             } catch(Exception e) {                                                 LOG.warn("Unexpected exception", e);                                             }                                         }                                     };                                 snapInProcess.start();                             }                             logCount = 0;                         }                     } else if (toFlush.isEmpty()) {                         // optimization for read heavy workloads                         // iff this is a read, and there are no pending                         // flushes (writes), then just pass this to the next                         // processor                         if (nextProcessor != null) {                             nextProcessor.processRequest(si); //继续调用下一个处理器来处理请求                             if (nextProcessor instanceof Flushable) {                                 ((Flushable)nextProcessor).flush();                             }                         }                         continue;                     }                     toFlush.add(si);                     if (toFlush.size() > 1000) {                         flush(toFlush);                     }                 }             }         } catch (Throwable t) {             handleException(this.getName(), t);         } finally{             running = false;         }         LOG.info("SyncRequestProcessor exited!");     }

FinalRequestProcessor. processRequest

这个方法就是咱们在课堂上分析到的方法了，FinalRequestProcessor.processRequest方法并根据Request对象中的操做更新内存中Session信息或者znode数据。

这块代码有小300多行，就不所有贴出来了，咱们直接定位到关键代码，根据客户端的OP类型找到以下的代码

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

case OpCode.exists: {                 lastOp = "EXIS";                 // TODO we need to figure out the security requirement for this!                 ExistsRequest existsRequest = new ExistsRequest();                 //反序列化  (将ByteBuffer反序列化成为ExitsRequest.这个就是咱们在客户端发起请求的时候传递过来的Request对象                 ByteBufferInputStream.byteBuffer2Record(request.request,                         existsRequest);                 String path = existsRequest.getPath(); //获得请求的路径                 if (path.indexOf('\0') != -1) {                     throw new KeeperException.BadArgumentsException();                 }                 //终于找到一个很关键的代码，判断请求的getWatch是否存在，若是存在，则传递cnxn（servercnxn） //对于exists请求，须要监听data变化事件，添加watcher                  Stat stat = zks.getZKDatabase().statNode(path, existsRequest.getWatch() ? cnxn : null);                 rsp = new ExistsResponse(stat); //在服务端内存数据库中根据路径获得结果进行组装，设置为ExistsResponse                 break;             }

　　

statNode这个方法作了什么？

1 2 3

public Stat statNode(String path, ServerCnxn serverCnxn) throws KeeperException.NoNodeException {     return dataTree.statNode(path, serverCnxn); }

一路向下，在下面这个方法中，讲ServerCnxn向上转型为Watcher了。 由于ServerCnxn实现了Watcher接口

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

public Stat statNode(String path, Watcher watcher)             throws KeeperException.NoNodeException {         Stat stat = new Stat();         DataNode n = nodes.get(path); //得到节点数据         if (watcher != null) { //若是watcher不为空，则讲当前的watcher和path进行绑定             dataWatches.addWatch(path, watcher);         }         if (n == null) {             throw new KeeperException.NoNodeException();         }         synchronized (n) {             n.copyStat(stat);             return stat;         }     }

WatchManager.addWatch(path, watcher);

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

synchronized void addWatch(String path, Watcher watcher) {         HashSet<Watcher> list = watchTable.get(path);  //判断watcherTable中是否存在当前路径对应的watcher         if (list == null) { //不存在则主动添加             // don't waste memory if there are few watches on a node             // rehash when the 4th entry is added, doubling size thereafter             // seems like a good compromise             list = new HashSet<Watcher>(4); // 新生成watcher集合             watchTable.put(path, list);         }         list.add(watcher); //添加到watcher表           HashSet<String> paths = watch2Paths.get(watcher);         if (paths == null) {             // cnxns typically have many watches, so use default cap here             paths = new HashSet<String>();             watch2Paths.put(watcher, paths); // 设置watcher到节点路径的映射         }         paths.add(path);  // 将路径添加至paths集合     }

　　

其大体流程以下

①经过传入的路径（节点路径）从watchTable获取相应的观察者集合，进入②

② 判断①中的观察者是否为空，若为空，则进入③，不然，进入④

③ 新生成观察者集合，并将路径路径和此集合添加至watchTable中，进入④

④将传输的观察者添加至观察者集合，即完成了路径和观察者添加至watchTable的步骤，进入⑤

⑤经过传入的观察者从watch2Paths中获取相应的路径集合，进入⑥

⑥ 判断路径集合是否为空，若为空，则进入⑦，不然，进入⑧

⑦ 新生成路径集合，并将观察者和路径添加至watch2Paths中，进入⑧

⑧将传入的路径（节点路径）添加至路径集合，即完成了路径和观察者添加至watch2Paths步骤的

总结
调用关系链以下

图文里的技术如何学习，有没有免费资料？

对Java技术，架构技术感兴趣的同窗，欢迎加QQ群619881427，一块儿学习，相互讨论。

群内已经有小伙伴将知识体系整理好（源码，笔记，PPT，学习视频），欢迎加群免费领取。

分享给喜欢Java的，喜欢编程，有梦想成为架构师的程序员们，但愿可以帮助到大家。

不是Java的程序员也不要紧，帮忙转发给更多朋友！谢谢。

一个分享小技巧点击阅读原文也。能够轻松到电子杂志学习资料哦！