Zookeeper的RPC框架

285572001

　　org.apache.zookeeper.ClientCnxn.java
　　该类管理客户端到HBase集群的socket I/O
　　所有需要通过socket发送的消息全部打包成Packet对象，然后放到ClentCnxn的outgoingQueue（LinkedList<Packet>）中，对outgoingQueue的操作需要同步控制。需要接受的消息也会被打包成Packet对象，放入pendingQueue（LinkedList<Packet>）中，等待一个答复。下面介绍一下设计的其他类及各类的使用方法及处理逻辑。
　　(一)Packet.java
　　



static class Packet {
RequestHeader header;
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;
ReplyHeader replyHeader;
Record request;
Record response;
boolean finished;
AsyncCallback cb;
Object ctx;
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;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("clientPath:" + clientPath);
sb.append(" serverPath:" + serverPath);
sb.append(" finished:" + finished);
sb.append(" header:: " + header);
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+", " ");
}
}

　　
　　
　　
　　该类含多个成员变量，主要的逻辑集中在构造函数中，若传入的字节数组（bb）内容非空，就将输入的"请求头（RequestHeader）"和“请求内容”（如：ExistRequest）转换成字节数组放入ByteBuffer中。其toString（）方法将所有的字段输出，并最后替换掉回车与换行之间的部分成空格。
　　1)RequestHeader.java
　　
　　在RequestHeader中有两个整型变量，Xid和type，type标志着操作类型，如：ping，auth，详见OpCode接口中的静态常量。
　　
　　Xid是一个操作序号，针对非ping和auth的操作，对于特定的ClientCnxn每次都会将该序号加1，然后传递给RequestHeader。
　　2)ReplyHeader extends record
　　在PendingQueue接收返回数据，
　　3)Request extends record
　　记录请求内容
　　4)Response extends record
　　 记录回应内容