Zookeeper 3.3.3消息序列化格式解析

朱雀 2022-06-18 02:54 207阅读 0赞

Zookeeper 客户端使用Packet来管理消息，. Packet管理 请求包头(RequestHeader)和请求包体(Record的不同实现，类似于ExistsRequest在调用zookeeper.exists时发送)，应答包头(ReplyHeader)和应答包体(Record的不同实现，类似于ExistsResponse). 与服务器进行通信时，依靠自己实现序列化方式来序列化请求包头以及请求包体，并且反序列化应答包头以及应答包体。

## 请求包序列化格式： ##

第一部分：4个字节表示整个包的长度(包头+包体)

第二部分：关注RequestHeader 类,Header包头,总共8个字节，前4个字节表示客户端生成的xid，用于跟踪包，采用自增方式生成的，需要服务器在响应时返回，后四个字节表示本包的操作类型：即ZooDefs.OpCode，它是一个整数值，分别代表"notification", "create","delete", "exists","getData", "setData", "getACL","setACL","getChildren", "getChildren2","getMaxChildren", "setMaxChildren", "ping"。getChildren和getChildren2操作的区别在于是否返回Stat统计数据，返回统计数据Stat的操作是getChildren2。

第三部分：关注Record接口的任何实现类，optional,因此长度字节数不固定,可以没有这一部分，例如ping操作,检测心跳不需要任何其他额外信息，只发Header,更多时候第三部分是存在的，根据第二部分的操作类型，可能需要提供一个操作的具体信息，例如操作路径path,是否watch等等，根据包体自己的方式来序列化。例如ExistsRequest

public void serialize(OutputArchive a_, String tag) throwsjava.io.IOException {
    
       a_.startRecord(this,tag);
    
       a_.writeString(path,"path");
    
       a_.writeBool(watch,"watch");
    
       a_.endRecord(this,tag);
    
      }

对于SetData操作来说，它包含有数据以及版本号：

public void serialize(OutputArchive a_, String tag) throwsjava.io.IOException {
    
       a_.startRecord(this,tag);
    
       a_.writeString(path,"path");
    
       a_.writeBuffer(data,"data");
    
       a_.writeInt(version,"version");
    
       a_.endRecord(this,tag);
    
      }

![0_13161946216NUT.gif][]

## 应答包序列化格式： ##

和请求包格式差不多，也分为三个部分

第一部分：4个字节表示整个应答包的长度(包头+包体)

第二部分：关注ReplyHeader类，16个字节长度，前面4个字节是xid,中间8个字节是zxid,表示事务id,最后4个字节表示err信息.

public void deserialize(InputArchive a_, String tag) throwsjava.io.IOException {
    
       a_.startRecord(tag);
    
       xid=a_.readInt("xid");
    
       zxid=a_.readLong("zxid");
    
       err=a_.readInt("err");
    
       a_.endRecord(tag);
    
    }

Err的值请关注Code类

第三部分：关注Record接口的任何实现类，例如对于ExistResponse。

public void deserialize(InputArchive a_, String tag) throwsjava.io.IOException {
    
       a_.startRecord(tag);
    
       stat= new org.apache.zookeeper.data.Stat();
    
       a_.readRecord(stat,"stat");
    
       a_.endRecord(tag);
    
    }

包体包含一个stat的统计信息。

![0_13161946117dPI.gif][]

## 客户端和服务器在交互时，有几个特殊的xid，分别的表示含义如下：  Request: Xid: -8表示reconnect时重新设置watches, -2表示ping 包，-4表示auth  Response: xid -2 表示ping的响应，-4 表示auth的响应，-1表示本次响应时有watch事件发生 ##

## 注： ##

客户端两个线程：一个发送线程，一个事件线程，发送线程从队列中依次发送请求，并接收响应，客户端生成xid，服务器返回这个xid，通过对比xid来保证接收到的消息有序，如果不是有序的，那么表示有消息丢失，丢失后，客户端会重新使用sessionid进行再一次连接服务器，每一次连接timeout或者readtimeout或者消息丢失都会导致之前注册的watch被保留，同步方法调用会抛出异常，异步调用方法会导致processResult调用，所有等待发送到服务器的outgoingQueue和已经发送去服务器等待应答的pendingQueue都将被clear。当session失效，那么这两个线程都会退出，需要重新创建一个zookeeper实例才能重新进行连接。

[0_13161946216NUT.gif]: /images/20220618/c91246b625474faaa5c923f24c0b18fb.png
[0_13161946117dPI.gif]: /images/20220618/ab68b73f0d834c10bcb863c64a6f9b86.png