zookeeper(4) 网络

时间 2019-12-14

原文原文链接

　　zookeeper底层经过NIO进行网络传输，若是对NIO不是很熟悉，先参见java NIO。下面咱们来逐步实现基于NIO的zookeeper实现，首先咱们要定义应用层的通讯协议。html

传输协议java

　　请求协议格式：缓存

	字段	长度	说明
	len	4	请求长度，解决半包粘包
请求头	xid	4	请求在客户端的id，用于惟一标识和保证响应顺序
请求头	type	4	请求类型code
请求体	request		不一样的请求类型，有不一样的请求体结构

　　响应协议格式：服务器

	字段	长度	解释
	len	4	响应长度，解决半包粘包
响应头	xid	4	包序号，惟一标识一个包，经过该标识找到对应的客户端Packet对象
	zxid	8	服务端事务处理id
	err	4	返回状态code
响应体	response		不一样的响应类型有不一样的响应体

　　请求和响应体类型：网络

List<String> childrensession

请求code	请求头类型	说明	请求格式	字段说明	响应格式	字段说明
0	notification
1	create	建立节点	String path	路径	String path	路径
			byte[] data	节点值
			List<ACL> acl	acl值
			int flags	节点类型
2	delete	删除节点	String path	路径	无
2	delete	删除节点	int version	版本	无
3	exists	是否存在指定节点	String path	路径	Stat stat	节点状态
3	exists	是否存在指定节点	boolean watch	是否有watch	Stat stat	节点状态
4	getData	获取节点数据	String path	路径	Stat stat	节点状态
4	getData	获取节点数据	boolean watch	是否有watch	byte[] data	节点数据
5	setData	设置节点数据	String path	路径	Stat stat	节点状态
			byte[] data	数据
			int version	版本
6	getACL	获取节点权限	String path	路径	List<ACL> acl	权限
6	getACL	获取节点权限	String path	路径	Stat stat	节点状态
7	setACL	设置节点权限	String path	路径	Stat stat	节点状态
			List<ACL> acl	权限
			int version	版本
8	getChildren	获取子节点	String path	路径	List<String> children	子节点名
8	getChildren	获取子节点	boolean watch	是否有watch	List<String> children	子节点名
9	sync	同步路径	String path	路径	String path	路径
11	ping
12	getChildren2
100	auth
101	setWatches
-10	createSession
-11	closeSession
－1	error

　　1.将请求封装成Packet对象放入outgoingQueue队列中。并发

　　2.有一个收发送线程会从outgoingQueue队列中取出一个可发送的Packet对象，并发送序列化信息，而后把该Packet放入到pendingQueue队列中。socket

　　3.收发线程会接收服务端响应，反序列号出结果数据，而后在pendingQueue中找到对应的Packet，设置结果，将Packet放入到waitingEvents队列中。ide

　　4.有一个事件线程，会不断从waitingEvents队列中取出一个Packet，并调用响应的callback方法。函数

发送packet包

Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
            Record response, AsyncCallback cb, String clientPath,
            String serverPath, Object ctx, WatchRegistration watchRegistration)
    {
        
        Packet packet = null;
        synchronized (outgoingQueue) {
            //设置一个全局惟一的id，做为数据包的id
            if (h.getType() != OpCode.ping && h.getType() != OpCode.auth) {
                h.setXid(getXid());
            }
            //将请求头，请求体，返回结果，watcher等封装成数据包。
            packet = new Packet(h, r, request, response, null,
                    watchRegistration);
            packet.cb = cb;
            packet.ctx = ctx;
            packet.clientPath = clientPath;
            packet.serverPath = serverPath;
            //将数据包添加到outgoing队列中。
            outgoingQueue.add(packet);
        }
        sendThread.wakeup();
        return packet;
    }

View Code

发送线程主流程（ClientCnxn.SendThread.run）：

class SendThread extends Thread {
        SelectionKey sockKey;
        private final Selector selector = Selector.open();
        public void run() {
            while (zooKeeper.state.isAlive()) {
                //创建链接
                startConnect();
                //获取注册通道
                selector.select(1000);
                Set<SelectionKey> selected;
                synchronized (this) {
                    selected = selector.selectedKeys();
                }
                for (SelectionKey k : selected) {
                    SocketChannel sc = ((SocketChannel) k.channel());
                    if ((k.readyOps() & SelectionKey.OP_CONNECT) != 0) {
                        //创建链接
                        if (sc.finishConnect()) {
                            primeConnection(k);
                        }
                        //读写数据
                    } else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
                        doIO();
                    }
                }
            }
            try {
                selector.close();
            } catch (IOException e) {
                LOG.warn("Ignoring exception during selector close", e);
            }
        }

      //经过nio创建链接
        private void startConnect() throws IOException {
            //从服务器列表中获取一台服务器地址
            InetSocketAddress addr = serverAddrs.get(nextAddrToTry);
            nextAddrToTry++;
            if (nextAddrToTry == serverAddrs.size()) {
                nextAddrToTry = 0;
            }
            //经过nio注册
            SocketChannel sock;
            sock = SocketChannel.open();
            sock.configureBlocking(false);
            sock.socket().setSoLinger(false, -1);
            sock.socket().setTcpNoDelay(true);
            try {
                sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
            } catch (IOException e) {
                sock.close();
                throw e;
            }
            //初始化缓存
            lenBuffer.clear();
            incomingBuffer = lenBuffer;
        }
}

View Code

处理读写主流程,主要是nio操做（ClientCnxn.SendThread.doIO）：

boolean doIO() throws InterruptedException, IOException {
            boolean packetReceived = false;
            //获取socketchannel
            SocketChannel sock = (SocketChannel) sockKey.channel();
            //若是可读
            if (sockKey.isReadable()) {
                //读取数据到缓存中
                int rc = sock.read(incomingBuffer);
                //直到缓存读满
                if (!incomingBuffer.hasRemaining()) {
                    //重置缓存
                    incomingBuffer.flip();
                    //若是读取的是长度信息，读取长度信息，而且分配相应缓存
                    if (incomingBuffer == lenBuffer) {
                        int len = incomingBuffer.getInt();
                        incomingBuffer = ByteBuffer.allocate(len);
                    } else if (!initialized) {
                      //若是是connect命令的返回值，获取session，timeout等相关信息
                        readConnectResult();
                        enableRead();
                        lenBuffer.clear();
                        //重置缓存
                        incomingBuffer = lenBuffer;
                        initialized = true;
                    } else {
                        //读取数据内容
                        readResponse();
                        //重置缓存
                        lenBuffer.clear();
                        incomingBuffer = lenBuffer;
                        packetReceived = true;
                    }
                }
            }
            //若是是写
            if (sockKey.isWritable()) {
                synchronized (outgoingQueue) {
                    if (!outgoingQueue.isEmpty()) {
                        //从outgoingQueue队列中拿数据包写入通道
                        ByteBuffer pbb = outgoingQueue.getFirst().bb;
                        sock.write(pbb);
                        if (!pbb.hasRemaining()) {
                            sentCount++;
                            Packet p = outgoingQueue.removeFirst();
                            if (p.header != null
                                    && p.header.getType() != OpCode.ping
                                    && p.header.getType() != OpCode.auth) {
                                pendingQueue.add(p);
                            }
                        }
                    }
                }
            }
            if (outgoingQueue.isEmpty()) {
                disableWrite();
            } else {
                enableWrite();
            }
            return packetReceived;
        }

View Code

处理返回结果，xid＝－2为ping命令的返回结果；xid＝-4为auth命令；xid＝－1为服务器发送的notification；其余命令返回结果。

void readResponse() throws IOException {
            //对返回数据进行反序列化
            ByteBufferInputStream bbis = new ByteBufferInputStream(
                    incomingBuffer);
            BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
            ReplyHeader replyHdr = new ReplyHeader();
            replyHdr.deserialize(bbia, "header");
            //根据返回头信息，封装想要的事件，放入事件队列中，交给eventthread处理
//向消息队列放入验证失败消息
            if (replyHdr.getXid() == -4) {
                 // -4 is the xid for AuthPacket               
                if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
                    zooKeeper.state = States.AUTH_FAILED;                    
                    eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None, 
                            Watcher.Event.KeeperState.AuthFailed, null) );                                        
                }
                return;
            }
//
            if (replyHdr.getXid() == -1) {
                WatcherEvent event = new WatcherEvent();
                event.deserialize(bbia, "response");
                if (chrootPath != null) {
                    String serverPath = event.getPath();
                    if(serverPath.compareTo(chrootPath)==0)
                        event.setPath("/");
                    else
                        event.setPath(serverPath.substring(chrootPath.length()));
                }
                WatchedEvent we = new WatchedEvent(event);
                eventThread.queueEvent( we );
                return;
            }
//反序列化返回结果
            Packet packet = null;
            synchronized (pendingQueue) {
                packet = pendingQueue.remove();
            }
            try {
                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");
                }
            } finally {
                finishPacket(packet);
            }
        }

View Code

private void finishPacket(Packet p) {
        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);
        }
    }

View Code

事件线程主要是处理回调函数（ClientCnxn.EventThread.run）：

public void run() {
           try {
              isRunning = true;
              while (true) {
                  //从队列中获取事件
                 Object event = waitingEvents.take();
                 if (event == eventOfDeath) {
                    wasKilled = true;
                 } else {
                     //处理事件
                    processEvent(event);
                 }
                 if (wasKilled)
                    synchronized (waitingEvents) {
                       if (waitingEvents.isEmpty()) {
                          isRunning = false;
                          break;
                       }
                    }
              }
           } catch (InterruptedException e) {
              LOG.error("Event thread exiting due to interruption", e);
           }
            LOG.info("EventThread shut down");
        }

View Code

处理回调函数和watcher

private void processEvent(Object event) {
          try {
              //处理watcher
              if (event instanceof WatcherSetEventPair) {
                  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 {
                  //
                  Packet p = (Packet) event;
                  int rc = 0;
                  String clientPath = p.clientPath;
                  if (p.replyHeader.getErr() != 0) {
                      rc = p.replyHeader.getErr();
                  }
                  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);
          }
       }

View Code

Packet结构

　　包，ClientCnxn内部管理请求内容的模块。由如下几个模块组成：

　　1.RequestHeader header 请求头

　　2.Record request 请求内容

　　3.ByteBuffer bb 实际须要发送的请求内容。

　　4.ReplyHeader replyHeader 响应头

　　5.Record response 响应内容

　　6.String clientPath

　　7.String serverPath

　　8.boolean finished

　　9.AsyncCallback cb

　　10.Object ctx

　　11.WatchRegistration watchRegistration

Packet(RequestHeader header, ReplyHeader replyHeader, Record record,
                Record response, ByteBuffer bb,
                WatchRegistration watchRegistration) {
            this.header = header;
            this.replyHeader = replyHeader;
            this.request = record;
            this.response = response;
            if (bb != null) {
                this.bb = bb;
            } else {
                try {
                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
                    BinaryOutputArchive boa = BinaryOutputArchive
                            .getArchive(baos);
                    boa.writeInt(-1, "len"); // We'll fill this in later
                    header.serialize(boa, "header");
                    if (record != null) {
                        record.serialize(boa, "request");
                    }
                    baos.close();
                    this.bb = ByteBuffer.wrap(baos.toByteArray());
                    this.bb.putInt(this.bb.capacity() - 4);
                    this.bb.rewind();
                } catch (IOException e) {
                    LOG.warn("Ignoring unexpected exception", e);
                }
            }
            this.watchRegistration = watchRegistration;
        }