深入浅出Zookeeper之四Create请求和处理
客户端接口[*]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);
页:
[1]