Zookeeper(八)序列化与协议
一 序列化与反序列化
对于一个网络通信,首先需要解决的就是对数据的序列化和反序列化处理,在ZooKeeper中,使用了Jute这一序列化组件来进行数据的序列化和反序列化操作。
1.1 Jute序列化工具
Zookeeper的客户端与服务端之间会进行一系列的网络通信来实现数据传输,Zookeeper使用Jute组件来完成数据的序列化和反序列化操作,其用于Zookeeper进行网络数据传输和本地磁盘数据存储的序列化和反序列化工作。 实体类要使用Jute进行序列化和反序列化步骤:
- 1.需要实现Record接口的serialize和deserialize方法;
- 2.构建一个序列化器BinaryOutputArchive;
- 3.序列化:调用实体类的serialize方法,将对象序列化到指定的tag中去,比如这里将对象序列化到header中;
- 4.反序列化:调用实体类的deserialize方法,从指定的tag中反序列化出数据内容。
package com.shu.jute;
import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.jute.Record;
/**
* @author 31380
* @description MockReHeader
* @create 2024/3/21 14:10
*/
public class MockReHeader implements Record {
private long sessionId;
private String type;
public MockReHeader() {}
public MockReHeader(long sessionId, String type) {
this.sessionId = sessionId;
this.type = type;
}
public void setSessionId(long sessionId) {
this.sessionId = sessionId;
}
public void setType(String type) {
this.type = type;
}
public long getSessionId() {
return sessionId;
}
public String getType() {
return type;
}
public void serialize(OutputArchive outputArchive, String tag) throws java.io.IOException {
outputArchive.startRecord(this, tag);
outputArchive.writeLong(sessionId, "sessionId");
outputArchive.writeString(type, "type");
outputArchive.endRecord(this, tag);
}
public void deserialize(InputArchive inputArchive, String tag) throws java.io.IOException {
inputArchive.startRecord(tag);
this.sessionId = inputArchive.readLong("sessionId");
this.type = inputArchive.readString("type");
inputArchive.endRecord(tag);
}
@Override
public String toString() {
return "sessionId = " + sessionId + ", type = " + type;
}
}
package com.shu.jute;
import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.zookeeper.server.ByteBufferInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
/**
* @author 31380
* @description
* @create 2024/3/21 14:11
*/
public class JuteTest {
public static void main(String[] args) throws IOException {
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
BinaryOutputArchive binaryOutputArchive = BinaryOutputArchive.getArchive(byteArrayOutputStream);
new MockReHeader(0x3421eccb92a34el, "ping").serialize(binaryOutputArchive, "header");
ByteBuffer byteBuffer = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
ByteBufferInputStream byteBufferInputStream = new ByteBufferInputStream(byteBuffer);
BinaryInputArchive binaryInputArchive = BinaryInputArchive.getArchive(byteBufferInputStream);
MockReHeader mockReHeader = new MockReHeader();
System.out.println(mockReHeader);
mockReHeader.deserialize(binaryInputArchive, "header");
System.out.println(mockReHeader);
byteBufferInputStream.close();
byteArrayOutputStream.close();
}
}
1.1 Recor接口
Zookeeper中所需要进行网络传输或是本地磁盘存储的类型定义,都实现了该接口,是Jute序列化的核心。Record定义了两个基本的方法,分别是serialize和deserialize,分别用于序列化和反序列化。其中archive是底层真正的序列化器和反序列化器,并且每个archive中可以包含对多个对象的序列化和反序列化,因此两个接口中都标记了参数tag,用于序列化器和反序列化器标识对象自己的标记。
1.2 OutputArchive和InputArchive
- OutputArchive和InputArchive分别是Jute底层的序列化器和反序列化器定义。有BinaryOutputArchive/BinaryInputArchive、CsvOutputArchive/CsvInputArchive和XmlOutputArchive/XmlInputArchive三种实现,无论哪种实现都是基于OutputStream和InputStream进行操作。
- BinaryOutputArchive对数据对象的序列化和反序列化,主要用于进行网络传输和本地磁盘的存储,是Zookeeper底层最主要的序列化方式。CsvOutputArchive对数据的序列化,更多的是方便数据的可视化展示,因此被用在toString方法中。XmlOutputArchive则是为了将数据对象以xml格式保存和还原,但目前在Zookeeper中基本没使用到。
注意:在最新版本的ZooKeeper中,底层依然使用了Jute这个古老的,并且似乎没有更多其他系统在使用的序列化组件。
二 通信协议
基于TCP/IP协议,ZooKeeper实现了自己的通信协议来完成客户端与服务端、服务端与服务端之间的网络通信。ZooKeeper通信协议整体上的设计非常简单,对于请求,主要包含请求头和请求体,而对于响应,则主要包含响应头和响应体
2.1 请求部分
我们将从请求头和请求体两方面分别解析ZooKeeper请求的协议设计
2.1.1 请求头
class RequestHeader {
int xid;
int type;
}
从zookeeper.jute中可知RequestHeader包含了xid和type,xid用于记录客户端请求发起的先后序号,用来确保单个客户端请求的响应顺序,type代表请求的操作类型,如创建节点(OpCode.create)、删除节点(OpCode.delete)、获取节点数据(OpCode.getData)。
2.2.2 请求体
协议的请求体部分是指请求的主体内容部分,包含了请求的所有操作内容。
ConnectRequest:会话创建
class ConnectRequest {
int protocolVersion;
long lastZxidSeen;
int timeOut;
long sessionId;
buffer passwd;
}
Zookeeper客户端和服务器在创建会话时,会发送ConnectRequest请求,该请求包含协议版本号protocolVersion、最近一次接收到服务器ZXID lastZxidSeen、会话超时时间timeOut、会话标识sessionId和会话密码passwd。
GetDataRequest:获取节点数据
class GetDataRequest {
ustring path;
boolean watch;
}
ZooKeeper客户端在向服务器发送获取节点数据请求的时候,会发送GetDataRequest请求,该请求体中包含了数据节点的节点路径path和是否注册Watcher的标识watch
SetDataRequest:更新节点数据
class SetDataRequest {
ustring path;
buffer data;
int version;
}
ZooKeeper客户端在向服务器发送更新节点数据请求的时候,会发送SetDataRequest请求,该请求体中包含了数据节点的节点路径path、数据内容data和节点数据的期望版本号version
2.1.3 案例分析
- 发出请求获取数据
我们获取到了ZooKeeper客户端请求发出后,在TCP层数据传输的十六进制表示,其中带下划线的部分就是对应的GetDataRequest请求,即[00,00,00,1d,00,00,00,01,00,00,00,04,00,00,00,10,2f,24,37,5f,32,5f,34,2f,67,65,74,5f,64,61,74,61,01],GetDataRequest请求的完整协议定义,我们来分析下这个十六进制字节数组的含义
2.2 响应部分
获取节点数据响应的完整协议定义
2.2.1 响应头
class ReplyHeader {
int xid;
long zxid;
int err;
}
xid与请求头中的xid一致,zxid表示Zookeeper服务器上当前最新的事务ID,err则是一个错误码,表示当请求处理过程出现异常情况时,就会在错误码中标识出来,常见的包括处理成功(Code.OK)、节点不存在(Code.NONODE)、没有权限(Code.NOAUTH)。
2.2.2 响应内容
协议的响应主体内容部分,包含了响应的所有数据,不同的响应类型请求体不同。
ConnectResponse:会话创建
class ConnectResponse {
int protocolVersion;
int timeOut;
long sessionId;
buffer passwd;
}
针对客户端的会话创建请求,服务端会返回客户端一个ConnectResponse响应,该响应体中包含了协议的版本号protocolVersion、会话的超时时间timeOut、会话标识sessionId和会话密码passwd
GetDataResponse:获取节点数据
class GetDataResponse {
buffer data;
org.apache.zookeeper.data.Stat stat;
}
针对客户端的获取节点数据请求,服务端会返回客户端一个GetDataResponse响应,该响应体中包含了数据节点的数据内容data和节点状态stat
SetDataResponse:更新节点数据
class SetDataResponse {
org.apache.zookeeper.data.Stat stat;
}
针对客户端的更新节点数据请求,服务端会返回客户端一个SetDataResponse响应,该响应体中包含了最新的节点状态stat
2.2.3 案例分析
我们获取到了ZooKeeper服务端响应发出之后,在TCP层数据传输的十六进制表示,其中带下划线的部分就是对应的GetDataResponse响应,即[00,00,00,63,00,00,00,05,00,00,00,00,00,00,00,04,00,00,00,00,00,00,00,0b,69,27,6d,5f,63,6f,6e,74,65,6e,74,00,00,00,00,00,00,00,04,00,00,00,00,00,00,00,04,00,00,01,43,67,bd,0e,08,00,00,01,43,67,bd,0e,08,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,00,0b,00,00,00,00,00,00,00,00,00,00,00,04]。