深入浅出Zookeeper之四Create请求和处理

zyllf2009

　　客户端接口
　　

• public String create(final String path,
  byte data[], List<ACL> acl,
  
• CreateMode createMode)
  
• throws KeeperException, InterruptedException
  
• {
  
• final String clientPath = path;
  
• PathUtils.validatePath(clientPath, createMode.isSequential());
  
• 
  
• final String serverPath = prependChroot(clientPath);
  
• //请求头
  
• RequestHeader h = new RequestHeader();
  
• h.setType(ZooDefs.OpCode.create);
  
• //请求体
  
• CreateRequest request = new CreateRequest();
  
• //CREATE请求需要server端响应
  
• CreateResponse response = new CreateResponse();
  
• request.setData(data);
  
• //node类型
  
• request.setFlags(createMode.toFlag());
  
• request.setPath(serverPath);
  
• if (acl != null && acl.size() ==
  0) {
  
• throw new KeeperException.InvalidACLException();
  
• }
  
• request.setAcl(acl);
  
• //同步提交
  
• ReplyHeader r = cnxn.submitRequest(h, request, response,
  null);
  
• //异常情况
  
• if (r.getErr() != 0) {
  
• throw KeeperException.create(KeeperException.Code.get(r.getErr()),
  
• clientPath);
  
• }
  
• //真实路径，对于SEQUENTIAL NODE后面会加上序号
  
• if (cnxn.chrootPath == null) {
  
• return response.getPath();
  
• } else {
  
• return response.getPath().substring(cnxn.chrootPath.length());
  
• }
  
• }
  

public String create(final String path, byte data[], List<ACL> acl,
CreateMode createMode)
throws KeeperException, InterruptedException
{
final String clientPath = path;
PathUtils.validatePath(clientPath, createMode.isSequential());
final String serverPath = prependChroot(clientPath);
//请求头
RequestHeader h = new RequestHeader();
h.setType(ZooDefs.OpCode.create);
//请求体
CreateRequest request = new CreateRequest();
//CREATE请求需要server端响应
CreateResponse response = new CreateResponse();
request.setData(data);
//node类型
request.setFlags(createMode.toFlag());
request.setPath(serverPath);
if (acl != null && acl.size() == 0) {
throw new KeeperException.InvalidACLException();
}
request.setAcl(acl);
//同步提交
ReplyHeader r = cnxn.submitRequest(h, request, response, null);
//异常情况
if (r.getErr() != 0) {
throw KeeperException.create(KeeperException.Code.get(r.getErr()),
clientPath);
}
//真实路径，对于SEQUENTIAL NODE后面会加上序号
if (cnxn.chrootPath == null) {
return response.getPath();
} else {
return response.getPath().substring(cnxn.chrootPath.length());
}
}

　　请求提交过程和之前的exists一样，都是通过sendthread写出去，都会进入pengding队列等待server端返回。
　　
　　server端处理也是一样，变化的只是RequestProcessor的业务逻辑。也就是说transport层是通用的，变化的是上层的业务层。
　　server端执行处理链，PrepRequestProcessor
　　

• switch (request.type) {
  
• case OpCode.create:
  
• //反序列化的对象
  
• CreateRequest createRequest = new CreateRequest();
  
• //zxid递增
  
• pRequest2Txn(request.type, zks.getNextZxid(), request, createRequest,
  true);
  
• break;
  
• ......
  
• request.zxid = zks.getZxid();
  
• nextProcessor.processRequest(request);