[Zookeeper学习笔记十]Zookeeper源代码分析之ClientCnxn数据序列化和反序列化

ClientCnxn是Zookeeper客户端和Zookeeper服务器端进行通信和事件通知处理的主要类，它内部包含两个类，1. SendThread 2. EventThread， SendThread负责客户端和服务器端的数据通信，也包括事件信息的传输，EventThread主要在客户端回调注册的Watchers进行通知处理

 

ClientCnxn构造方法

 

 

    /**
     * Creates a connection object. The actual network connect doesn't get
     * established until needed. The start() instance method must be called
     * subsequent to construction.
     *
     * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
     * @param hostProvider
     *                the list of ZooKeeper servers to connect to
     * @param sessionTimeout
     *                the timeout for connections.
     * @param zooKeeper
     *                the zookeeper object that this connection is related to.
     * @param watcher watcher for this connection
     * @param clientCnxnSocket
     *                the socket implementation used (e.g. NIO/Netty)
     * @param sessionId session id if re-establishing session
     * @param sessionPasswd session passwd if re-establishing session
     * @param canBeReadOnly
     *                whether the connection is allowed to go to read-only
     *                mode in case of partitioning
     * @throws IOException
     */
    public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
            ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
            long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
        this.zooKeeper = zooKeeper;
        this.watcher = watcher;
        this.sessionId = sessionId;
        this.sessionPasswd = sessionPasswd;
        this.sessionTimeout = sessionTimeout;
        this.hostProvider = hostProvider;
        this.chrootPath = chrootPath;
        //如果zookeeper集群有1000台，那么会话超时时间岂不是要设置的很大？因此，zookeeper一般不会很大，3台或者5台足亦
        connectTimeout = sessionTimeout / hostProvider.size();//链接超时时间是会话超时时间除以Zookeeper集群数
        //读超时是会话超时的2/3
        readTimeout = sessionTimeout * 2 / 3;
        readOnly = canBeReadOnly;
   
        //读线程，在ClientCnxnd的start方法中启动
        sendThread = new SendThread(clientCnxnSocket);
        //会话线程，在ClientCnxnd的start方法中启动
        eventThread = new EventThread();

    }

 

 对于SendThread数据传输线程包含两方面的内容，1是基于TCP/IP的Socket的数据传输，2.传输的数据内容。首先关注传输的数据内容，在TCP/IP传输的数据都是字节，因此，在SendThread发送数据之前，需要将要传输的数据结构进行序列化成字节流，服务器端会反序列化成相应的数据结构。当客户端收到服务器返回的字节流时，客户端将其反序列化为相应的数据结构。

 

首先看看数据的序列化和反序列化，接口定义：

package org.apache.jute //该借口不是Zookeeper原生提供的，是Apache的jute提供的
import java.io.IOException;

/**
 * Interface that is implemented by generated classes.
 * 
 */
public interface Record {
    public void serialize(OutputArchive archive, String tag) //序列化，tag在XmlInputArchive序列化器中，用作xml元素标签
        throws IOException;
    public void deserialize(InputArchive archive, String tag)//反序列话
        throws IOException;
}

 OutputArchive接口是数据结构序列化为字节流的字节流写入器，InputArchive接口是字节流反序列化为数据结构的字节流读取器。OutputArchive和InputArchive接口有三个成对使用的实现类

BinaryOutputArchive和BinaryInputArchive底层使用DataOutput和DataInput作为字节容器

 

XmlOutputArchive和XmlInputArchive底层使用PrintStream，XmlInputArchive使用Xml解析的方式得到相应的数据结构

CsvOutputArchive和CsvInputArchive底层使用PrintStream作为自己容器

 

Zookeeper客户端向服务器端发送请求，包含请求头和请求正文两部分，每个请求的请求头的类型都是一样的，而请求正文根据请求的不同，分为多种类型。

Zookeeper服务器端向客户端返回响应数据，包含响应头和响应正文两部分，每个响应的响应头的类型都是一样的，而响应正文根据请求的不同，分为多种类型。

 

请求头，各种请求正文，响应头和响应正文因为要在Socket上进行数据传输，所以它们应该都是可序列化和反序列话的，因此它们都是可序列化的

 

 

请求头：

public class RequestHeader implements Record {
  private int xid; //请求的事务id，具体的含义和功能接下来分析
  private int type; //请求类型？
  public RequestHeader() {
  }
  public RequestHeader(
        int xid,
        int type) {
    this.xid=xid;
    this.type=type;
  }
  public int getXid() {
    return xid;
  }
  public void setXid(int m_) {
    xid=m_;
  }
  public int getType() {
    return type;
  }
  public void setType(int m_) {
    type=m_;
  }
  //序列化操作，将xid和type序列化到OutputArchive中，
  public void serialize(OutputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(this,tag); //对于最常使用的BinaryOutputArchive，此方法空实现
    a_.writeInt(xid,"xid");
    a_.writeInt(type,"type");
    a_.endRecord(this,tag);//对于最常使用的BinaryOutputArchive，此方法空实现
  }
  //序列化操作，将xid和type反序列化
  public void deserialize(InputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(tag);//对于最常使用的BinaryInputArchive，此方法空实现
    xid=a_.readInt("xid");
    type=a_.readInt("type");
    a_.endRecord(tag);//对于最常使用的BinaryInputArchive，此方法空实现
}
  public String toString() {
    try {
      java.io.ByteArrayOutputStream s =
        new java.io.ByteArrayOutputStream();
      CsvOutputArchive a_ = 
        new CsvOutputArchive(s);
      a_.startRecord(this,"");//对于CsvOutputArchive，startRecord方法
    a_.writeInt(xid,"xid");
    a_.writeInt(type,"type");
      a_.endRecord(this,"");
      return new String(s.toByteArray(), "UTF-8");
    } catch (Throwable ex) {
      ex.printStackTrace();
    }
    return "ERROR";
  }
  public void write(java.io.DataOutput out) throws java.io.IOException {
    BinaryOutputArchive archive = new BinaryOutputArchive(out);
    serialize(archive, "");
  }
  public void readFields(java.io.DataInput in) throws java.io.IOException {
    BinaryInputArchive archive = new BinaryInputArchive(in);
    deserialize(archive, "");
  }
  public int compareTo (Object peer_) throws ClassCastException {
    if (!(peer_ instanceof RequestHeader)) {
      throw new ClassCastException("Comparing different types of records.");
    }
    RequestHeader peer = (RequestHeader) peer_;
    int ret = 0;
    ret = (xid == peer.xid)? 0 :((xid<peer.xid)?-1:1);
    if (ret != 0) return ret;
    ret = (type == peer.type)? 0 :((type<peer.type)?-1:1);
    if (ret != 0) return ret;
     return ret;
  }
  public boolean equals(Object peer_) {
    if (!(peer_ instanceof RequestHeader)) {
      return false;
    }
    if (peer_ == this) {
      return true;
    }
    RequestHeader peer = (RequestHeader) peer_;
    boolean ret = false;
    ret = (xid==peer.xid);
    if (!ret) return ret;
    ret = (type==peer.type);
    if (!ret) return ret;
     return ret;
  }
  public int hashCode() {
    int result = 17;
    int ret;
    ret = (int)xid;
    result = 37*result + ret;
    ret = (int)type;
    result = 37*result + ret;
    return result;
  }
  public static String signature() {
    return "LRequestHeader(ii)";
  }
}

 

请求正文有很多，比如

• 链接请求ConnectRequest
• 创建znode请求CreateRequest
• 节点是否存在请求ExistsRequest
• 删除znode请求DeleteRequest
• 获取child znodes请求GetChildrenRequest
• 设置znode数据SetDataRequest
• 事件WatcherEvent

以CreateRequest为例进行分析

 

public class CreateRequest implements Record {
  private String path; //创建znode节点的path
  private byte[] data; //创建znode节点时的节点数据
  private java.util.List<org.apache.zookeeper.data.ACL> acl; //创建znode节点时的ACL
  private int flags;//这个参数干啥的？
  public CreateRequest() {
  }
  public CreateRequest(
        String path,
        byte[] data,
        java.util.List<org.apache.zookeeper.data.ACL> acl,
        int flags) {
    this.path=path;
    this.data=data;
    this.acl=acl;
    this.flags=flags;
  }
  public String getPath() {
    return path;
  }
  public void setPath(String m_) {
    path=m_;
  }
  public byte[] getData() {
    return data;
  }
  public void setData(byte[] m_) {
    data=m_;
  }
  public java.util.List<org.apache.zookeeper.data.ACL> getAcl() {
    return acl;
  }
  public void setAcl(java.util.List<org.apache.zookeeper.data.ACL> m_) {
    acl=m_;
  }
  public int getFlags() {
    return flags;
  }
  public void setFlags(int m_) {
    flags=m_;
  }
  public void serialize(OutputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(this,tag);
    a_.writeString(path,"path");//写入path
    a_.writeBuffer(data,"data");//写入data字节数组
    {
      a_.startVector(acl,"acl");//写入acl，acl是List类型
      if (acl!= null) {        
          int len1 = acl.size();
          for(int vidx1 = 0; vidx1<len1; vidx1++) {
            org.apache.zookeeper.data.ACL e1 = (org.apache.zookeeper.data.ACL) acl.get(vidx1);
            a_.writeRecord(e1,"e1");//ACL也是一个Record
          }
      }
      a_.endVector(acl,"acl");
    }
    a_.writeInt(flags,"flags");//写入flags
    a_.endRecord(this,tag);
  }
  public void deserialize(InputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(tag);
    path=a_.readString("path");
    data=a_.readBuffer("data");
    {
      Index vidx1 = a_.startVector("acl");
      if (vidx1!= null) {          acl=new java.util.ArrayList<org.apache.zookeeper.data.ACL>();
          for (; !vidx1.done(); vidx1.incr()) {
    org.apache.zookeeper.data.ACL e1;
    e1= new org.apache.zookeeper.data.ACL();
    a_.readRecord(e1,"e1");
            acl.add(e1);
          }
      }
    a_.endVector("acl");
    }
    flags=a_.readInt("flags");
    a_.endRecord(tag);
}
  public String toString() {
    try {
      java.io.ByteArrayOutputStream s =
        new java.io.ByteArrayOutputStream();
      CsvOutputArchive a_ = 
        new CsvOutputArchive(s);
      a_.startRecord(this,"");
    a_.writeString(path,"path");
    a_.writeBuffer(data,"data");
    {
      a_.startVector(acl,"acl");
      if (acl!= null) {          int len1 = acl.size();
          for(int vidx1 = 0; vidx1<len1; vidx1++) {
            org.apache.zookeeper.data.ACL e1 = (org.apache.zookeeper.data.ACL) acl.get(vidx1);
    a_.writeRecord(e1,"e1");
          }
      }
      a_.endVector(acl,"acl");
    }
    a_.writeInt(flags,"flags");
      a_.endRecord(this,"");
      return new String(s.toByteArray(), "UTF-8");
    } catch (Throwable ex) {
      ex.printStackTrace();
    }
    return "ERROR";
  }
  public void write(java.io.DataOutput out) throws java.io.IOException {
    BinaryOutputArchive archive = new BinaryOutputArchive(out);
    serialize(archive, "");
  }
  public void readFields(java.io.DataInput in) throws java.io.IOException {
    BinaryInputArchive archive = new BinaryInputArchive(in);
    deserialize(archive, "");
  }
  public int compareTo (Object peer_) throws ClassCastException {
    throw new UnsupportedOperationException("comparing CreateRequest is unimplemented");
  }
  public boolean equals(Object peer_) {
    if (!(peer_ instanceof CreateRequest)) {
      return false;
    }
    if (peer_ == this) {
      return true;
    }
    CreateRequest peer = (CreateRequest) peer_;
    boolean ret = false;
    ret = path.equals(peer.path);
    if (!ret) return ret;
    ret = org.apache.jute.Utils.bufEquals(data,peer.data);
    if (!ret) return ret;
    ret = acl.equals(peer.acl);
    if (!ret) return ret;
    ret = (flags==peer.flags);
    if (!ret) return ret;
     return ret;
  }
  public int hashCode() {
    int result = 17;
    int ret;
    ret = path.hashCode();
    result = 37*result + ret;
    ret = java.util.Arrays.toString(data).hashCode();
    result = 37*result + ret;
    ret = acl.hashCode();
    result = 37*result + ret;
    ret = (int)flags;
    result = 37*result + ret;
    return result;
  }
  public static String signature() {
    return "LCreateRequest(sB[LACL(iLId(ss))]i)";
  }
}

 

 

 ConnectRequest的请求数据：

  private int protocolVersion;
  private long lastZxidSeen; //客户端保存的Zxid最近时间，zxid有什么用呢？
  private int timeOut;//会话超时时间
  private long sessionId;
  private byte[] passwd;

 

 

 

 ClientCnxn的内部类Packet类封装了请求头，响应头，请求正文和响应征正文

   static class Packet {
        RequestHeader requestHeader;//请求头

        ReplyHeader replyHeader; //响应头

        Record request;//请求正文

        Record response; //响应正文

        ByteBuffer bb;//上面四部分序列化的字节流

        /** Client's view of the path (may differ due to chroot) **/
        String clientPath;
        /** Servers's view of the path (may differ due to chroot) **/
        String serverPath;

        boolean finished;

        AsyncCallback cb;//异步请求的响应Callback

        Object ctx;

        WatchRegistration watchRegistration;

        public boolean readOnly;

        /** Convenience ctor */
        Packet(RequestHeader requestHeader, ReplyHeader replyHeader,
               Record request, Record response,
               WatchRegistration watchRegistration) {
            this(requestHeader, replyHeader, request, response,
                 watchRegistration, false);
        }

        Packet(RequestHeader requestHeader, ReplyHeader replyHeader,
               Record request, Record response,
               WatchRegistration watchRegistration, boolean readOnly) {

            this.requestHeader = requestHeader;
            this.replyHeader = replyHeader;
            this.request = request;
            this.response = response;
            this.readOnly = readOnly;
            this.watchRegistration = watchRegistration;
        }

        public void createBB() {
            try {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);//序列化字节流容器
                boa.writeInt(-1, "len"); // We'll fill this in later
                if (requestHeader != null) {
                    requestHeader.serialize(boa, "header");
                }
                if (request instanceof ConnectRequest) {
                    request.serialize(boa, "connect");
                    // append "am-I-allowed-to-be-readonly" flag
                    boa.writeBool(readOnly, "readOnly");
                } else if (request != null) {
                    request.serialize(boa, "request");
                }
                baos.close();
                this.bb = ByteBuffer.wrap(baos.toByteArray());//将字节流容器中的字节流复制给bb
                this.bb.putInt(this.bb.capacity() - 4);
                this.bb.rewind();
            } catch (IOException e) {
                LOG.warn("Ignoring unexpected exception", e);
            }
        }

        @Override
        public String toString() {
            StringBuilder sb = new StringBuilder();

            sb.append("clientPath:" + clientPath);
            sb.append(" serverPath:" + serverPath);
            sb.append(" finished:" + finished);

            sb.append(" header:: " + requestHeader);
            sb.append(" replyHeader:: " + replyHeader);
            sb.append(" request:: " + request);
            sb.append(" response:: " + response);

            // jute toString is horrible, remove unnecessary newlines
            return sb.toString().replaceAll("\r*\n+", " ");
        }
    }

    ClientCnxn类包含两个队列(LinkedList)，队列中的元素都是Packet类型，pengdingQueue表示请求已经发送，等待响应结果；outgoingQueue表示等待发送请求的请求序列

 

  

/**
     * These are the packets that have been sent and are waiting for a response.
     */
    private final LinkedList<Packet> pendingQueue = new LinkedList<Packet>();

    /**
     * These are the packets that need to be sent.
     */
    private final LinkedList<Packet> outgoingQueue = new LinkedList<Packet>();

 

 

 ClientCnxn的Socket的数据传输，将另外一篇进行单独分析