读《从Paxos到Zookeeper 分布式一致性原理与实践》笔记之通信协议

秒速五厘米 2022-07-13 00:30 174阅读 0赞

# 1.    Zookeeper技术内幕 #

## 1.1. 通信协议 ##

基于TCP/IP协议，zookeeper实现了自己的通信协议来完成客户端与服务端、服务端与服务端之间的网络通信。Zookeeper通信协议整体上的设计非常简单，对于请求，主要包含请求头和请求体，而对于响应，则主要包含响应头和响应体。

![Center][]

### 1.1.1.  协议解析：请求部分 ###

GetDataRequest“获取节点数据”请求的完整协议定义

<table> 
 <tbody> 
 <tr> 
 <td style="background:#C3BD96"> &nbsp; </td> 
 <td style="background:#C3BD96"> 请求长度 </td> 
 <td colspan="2" style="background:#C3BD96"> 请求头 </td> 
 <td colspan="3" style="background:#C3BD96"> 请求体 </td> 
 </tr> 
 <tr> 
 <td> 字节偏移量 </td> 
 <td> 0 - 3 </td> 
 <td colspan="2"> 4 - 11 </td> 
 <td colspan="3"> 12 - n </td> 
 </tr> 
 <tr> 
 <td> 4 - 7 </td> 
 <td> 8 - 11 </td> 
 <td> 12 - 15 </td> 
 <td> 16-(n-1) </td> 
 <td> n </td> 
 </tr> 
 <tr> 
 <td> 协议内容 </td> 
 <td> len </td> 
 <td> xid </td> 
 <td> type </td> 
 <td> len（数据体长度） </td> 
 <td> path </td> 
 <td> watch </td> 
 </tr> 
 </tbody> 
</table>

**请求头：RequestHeader**

public class RequestHeader implements Record {
      private int xid;
      private int type;
    }

xid用于记录请求发起的先后序号，用于确定单个客户端请求的响应顺序。Type代表请求的操作类型，常见的包括创建节点(OpCode.create)、删除节点(OpCode.delete)和获取节点数据(OpCode.getData)等。

**请求体：Request**

协议的请求体部分是指请求的主体内容部分，包含了请求的所有操作内容。不同的请求类型，其请求体部分的结构是不同的，下面以获取节点数据的请求体为例来对请求体进行分析。

public class GetDataRequest implements Record {
      private String path;
      private boolean watch;
    }

该请求体包含了数据节点的节点路径path和是否注册Watcher的标识watch。

序列化请求协议到缓存中：

static class Packet {
        RequestHeader requestHeader;
        ReplyHeader replyHeader;
        Record request;
        Record response;
        ByteBuffer bb;
        ......
        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 != null) {
                    request.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);
            }
        }
    }

### 1.1.2.  协议解析：响应部分 ###

以GetDataResponse“获取节点数据”响应为例，解析完整协议定义

<table> 
 <tbody> 
 <tr> 
 <td style="background:#C3BD96"> &nbsp; </td> 
 <td style="background:#C3BD96"> 请求长度 </td> 
 <td colspan="3" style="background:#C3BD96"> 响应头 </td> 
 <td colspan="3" style="background:#C3BD96"> 响应体 </td> 
 </tr> 
 <tr> 
 <td> 字节偏移量 </td> 
 <td> 0 - 3 </td> 
 <td colspan="3"> 4 - 19 </td> 
 <td colspan="3"> 20 - n </td> 
 </tr> 
 <tr> 
 <td> 4 - 7 </td> 
 <td> 8 - 15 </td> 
 <td> 16 - 19 </td> 
 <td> 20 - 23 </td> 
 <td> len位 </td> 
 <td> 68位 </td> 
 </tr> 
 <tr> 
 <td> 协议内容 </td> 
 <td> len </td> 
 <td> xid </td> 
 <td> zxid </td> 
 <td> err </td> 
 <td> len </td> 
 <td> data </td> 
 <td> Stat </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td style="background:#C3BD96"> 8 位 </td> 
 <td style="background:#C3BD96"> 8 位 </td> 
 <td style="background:#C3BD96"> 8 位 </td> 
 <td style="background:#C3BD96"> 8 位 </td> 
 <td style="background:#C3BD96"> 4 位 </td> 
 <td style="background:#C3BD96"> 4 位 </td> 
 <td style="background:#C3BD96"> 4 位 </td> 
 <td style="background:#C3BD96"> 8 位 </td> 
 <td style="background:#C3BD96"> 4 位 </td> 
 <td style="background:#C3BD96"> 4 位 </td> 
 <td style="background:#C3BD96"> 8位 </td> 
 </tr> 
 <tr> 
 <td> czxid </td> 
 <td> mzxid </td> 
 <td> ctime </td> 
 <td> mtime </td> 
 <td> version </td> 
 <td> cversion </td> 
 <td> aversion </td> 
 <td> ephemeral Owner </td> 
 <td> dataLength </td> 
 <td> numChildren </td> 
 <td> pzxid </td> 
 </tr> 
 </tbody> 
</table>

**响应头：ReplyHeader**

public class ReplyHeader implements Record {
      private int xid;
      private long zxid;
      private int err;
    }

Xid和上文中提到的请求头中的xid是一致的，响应中只是将请求的xid原值返回。Zxid代表zookeeper服务器上当前最新的事物Id。Err则是一个错误码，当请求处理过程中出现异常情况时，会在这个错误码中表示出来，常见的包括处理成功(Code.OK)、节点不存在(Code.NONODE)和没有权限(Code.NOAUTH)等。

**响应体：Request**

协议的响应体部分是指响应的主体内容部分，包含了相应的所有返回数据。不通的响应类型，其相应体部分的结构是不同的，下面以获取节点数据的响应体为例来对响应体进行分析。

public class GetDataResponse implements Record {
      private byte[] data;
      private org.apache.zookeeper.data.Stat stat;
    }
    public class Stat implements Record {
      private long czxid;
      private long mzxid;
      private long ctime;
      private long mtime;
      private int version;
      private int cversion;
      private int aversion;
      private long ephemeralOwner;
      private int dataLength;
      private int numChildren;
      private long pzxid;
    }

反序列化响应协议到缓存中：

class SendThread extends ZooKeeperThread {
        void readResponse(ByteBuffer incomingBuffer) throws IOException {
            ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
            BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
            ReplyHeader replyHdr = new ReplyHeader();
            replyHdr.deserialize(bbia, "header");
    ......
            Packet packet;
            synchronized (pendingQueue) {
                packet = pendingQueue.remove();
            }
            try {
                if (packet.requestHeader.getXid() != replyHdr.getXid()) {
                    packet.replyHeader.setErr(
                            KeeperException.Code.CONNECTIONLOSS.intValue());
                    throw new IOException("Xid out of order. Got Xid ");
                }
                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);
            }
        }
    }

[Center]: /images/20220713/449801c28d6044f799def74b1ec4cb28.png