一、前言
前面分析了SyncReqeustProcessor，接着分析请求处理链中最后的一个处理器FinalRequestProcessor。
二、FinalRequestProcessor源码分析
2.1 类的继承关系
public class FinalRequestProcessor implements RequestProcessor {}
说明：FinalRequestProcessor只实现了RequestProcessor接口，其需要实现processRequest方法和shutdown方法。
2.2 类的属性
public class FinalRequestProcessor implements RequestProcessor {
private static final Logger LOG = LoggerFactory.getLogger(FinalRequestProcessor.class);
// ZooKeeper服务器
ZooKeeperServer zks;
}
说明：其核心属性为zks，表示Zookeeper服务器，可以通过zks访问到Zookeeper内存数据库。
2.3 类的构造函数
public FinalRequestProcessor(ZooKeeperServer zks) {
this.zks = zks;
}
2.4 核心函数分析
1. processRequest
public void processRequest(Request request) {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing request:: " + request);
}
// request.addRQRec(">final");
long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
if (request.type == OpCode.ping) { // 请求类型为PING
traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
}
if (LOG.isTraceEnabled()) {
ZooTrace.logRequest(LOG, traceMask, 'E', request, "");
}
ProcessTxnResult rc = null;
synchronized (zks.outstandingChanges) { // 同步块
while (!zks.outstandingChanges.isEmpty()
&& zks.outstandingChanges.get(0).zxid <= request.zxid) { // outstandingChanges不为空且首个元素的zxid小于请求的zxid
// 移除首个元素
ChangeRecord cr = zks.outstandingChanges.remove(0);
if (cr.zxid < request.zxid) { // 若Record的zxid小于请求的zxid
LOG.warn("Zxid outstanding "
+ cr.zxid
+ " is less than current " + request.zxid);
}
if (zks.outstandingChangesForPath.get(cr.path) == cr) { // 根据路径得到Record并判断是否为cr
// 移除cr的路径对应的记录
zks.outstandingChangesForPath.remove(cr.path);
}
}
if (request.hdr != null) { // 请求头不为空
// 获取请求头
TxnHeader hdr = request.hdr;
// 获取请求事务
Record txn = request.txn;
// 处理事务
rc = zks.processTxn(hdr, txn);
}
// do not add non quorum packets to the queue.
if (Request.isQuorum(request.type)) { // 只将quorum包（事务性请求）添加进队列
zks.getZKDatabase().addCommittedProposal(request);
}
}
if (request.hdr != null && request.hdr.getType() == OpCode.closeSession) { // 请求头不为空并且请求类型为关闭会话
ServerCnxnFactory scxn = zks.getServerCnxnFactory();
// this might be possible since
// we might just be playing diffs from the leader
if (scxn != null && request.cnxn == null) { //
// calling this if we have the cnxn results in the client's
// close session response being lost - we've already closed
// the session/socket here before we can send the closeSession
// in the switch block below
// 关闭会话
scxn.closeSession(request.sessionId);
return;
}
}
if (request.cnxn == null) { // 请求的cnxn为空，直接返回
return;
}
ServerCnxn cnxn = request.cnxn;
String lastOp = "NA";
zks.decInProcess();
Code err = Code.OK;
Record rsp = null;
boolean closeSession = false;
try {
if (request.hdr != null && request.hdr.getType() == OpCode.error) {
throw KeeperException.create(KeeperException.Code.get((
(ErrorTxn) request.txn).getErr()));
}
KeeperException ke = request.getException();
if (ke != null && request.type != OpCode.multi) {
throw ke;
}
if (LOG.isDebugEnabled()) {
LOG.debug("{}",request);
}
switch (request.type) {
case OpCode.ping: { // PING请求
// 更新延迟
zks.serverStats().updateLatency(request.createTime);
lastOp = "PING";
// 更新响应的状态
cnxn.updateStatsForResponse(request.cxid, request.zxid, lastOp,
request.createTime, System.currentTimeMillis());
// 设置响应
cnxn.sendResponse(new ReplyHeader(-2,
zks.getZKDatabase().getDataTreeLastProcessedZxid(), 0), null, "response");
return;
}
case OpCode.createSession: { // 创建会话请求
// 更新延迟
zks.serverStats().updateLatency(request.createTime);
lastOp = "SESS";
// 更新响应的状态
cnxn.updateStatsForResponse(request.cxid, request.zxid, lastOp,
request.createTime, System.currentTimeMillis());
// 结束会话初始化
zks.finishSessionInit(request.cnxn, true);
return;
}
case OpCode.multi: { // 多重操作
lastOp = "MULT";
rsp = new MultiResponse() ;
for (ProcessTxnResult subTxnResult : rc.multiResult) { // 遍历多重操作结果
OpResult subResult ;
switch (subTxnResult.type) { // 确定每个操作类型
case OpCode.check: // 检查
subResult = new CheckResult();
break;
case OpCode.create: // 创建
subResult = new CreateResult(subTxnResult.path);
break;
case OpCode.delete: // 删除
subResult = new DeleteResult();
break;
case OpCode.setData: // 设置数据
subResult = new SetDataResult(subTxnResult.stat);
break;
case OpCode.error: // 错误
subResult = new ErrorResult(subTxnResult.err) ;
break;
default:
throw new IOException("Invalid type of op");
}
// 添加至响应结果集中
((MultiResponse)rsp).add(subResult);
}
break;
}
case OpCode.create: { // 创建
lastOp = "CREA";
// 创建响应
rsp = new CreateResponse(rc.path);
err = Code.get(rc.err);
break;
}
case OpCode.delete: { // 删除
lastOp = "DELE";
err = Code.get(rc.err);
break;
}
case OpCode.setData: { // 设置数据
lastOp = "SETD";
rsp = new SetDataResponse(rc.stat);
err = Code.get(rc.err);
break;
}
case OpCode.setACL: { // 设置ACL
lastOp = "SETA";
rsp = new SetACLResponse(rc.stat);
err = Code.get(rc.err);
break;
}
case OpCode.closeSession: { // 关闭会话
lastOp = "CLOS";
closeSession = true;
err = Code.get(rc.err);
break;
}
case OpCode.sync: { // 同步
lastOp = "SYNC";
SyncRequest syncRequest = new SyncRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
syncRequest);
rsp = new SyncResponse(syncRequest.getPath());
break;
}
case OpCode.check: { // 检查
lastOp = "CHEC";
rsp = new SetDataResponse(rc.stat);
err = Code.get(rc.err);
break;
}
case OpCode.exists: { // 存在性判断
lastOp = "EXIS";
// TODO we need to figure out the security requirement for this!
ExistsRequest existsRequest = new ExistsRequest();
// 将byteBuffer转化为Record
ByteBufferInputStream.byteBuffer2Record(request.request,
existsRequest);
String path = existsRequest.getPath();
if (path.indexOf('\0') != -1) {
throw new KeeperException.BadArgumentsException();
}
Stat stat = zks.getZKDatabase().statNode(path, existsRequest
.getWatch() ? cnxn : null);
rsp = new ExistsResponse(stat);
break;
}
case OpCode.getData: { // 获取数据
lastOp = "GETD";
GetDataRequest getDataRequest = new GetDataRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
getDataRequest);
DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath());
if (n == null) {
throw new KeeperException.NoNodeException();
}
Long aclL;
synchronized(n) {
aclL = n.acl;
}
PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().convertLong(aclL),
ZooDefs.Perms.READ,
request.authInfo);
Stat stat = new Stat();
byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
getDataRequest.getWatch() ? cnxn : null);
rsp = new GetDataResponse(b, stat);
break;
}
case OpCode.setWatches: { // 设置watch
lastOp = "SETW";
SetWatches setWatches = new SetWatches();
// XXX We really should NOT need this!!!!
request.request.rewind();
ByteBufferInputStream.byteBuffer2Record(request.request, setWatches);
long relativeZxid = setWatches.getRelativeZxid();
zks.getZKDatabase().setWatches(relativeZxid,
setWatches.getDataWatches(),
setWatches.getExistWatches(),
setWatches.getChildWatches(), cnxn);
break;
}
case OpCode.getACL: { // 获取ACL
lastOp = "GETA";
GetACLRequest getACLRequest = new GetACLRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
getACLRequest);
Stat stat = new Stat();
List<ACL> acl =
zks.getZKDatabase().getACL(getACLRequest.getPath(), stat);
rsp = new GetACLResponse(acl, stat);
break;
}
case OpCode.getChildren: { // 获取子节点
lastOp = "GETC";
GetChildrenRequest getChildrenRequest = new GetChildrenRequest();
ByteBufferInputStream.byteBuffer2Record(request.request,
getChildrenRequest);
DataNode n = zks.getZKDatabase().getNode(getChildrenRequest.getPath());
if (n == null) {
throw new KeeperException.NoNodeException();
}
Long aclG;
synchronized(n) {
aclG = n.acl;
}
PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().convertLong(aclG),
ZooDefs.Perms.READ,
request.authInfo);
List<String> children = zks.getZKDatabase().getChildren(
getChildrenRequest.getPath(), null, getChildrenRequest
.getWatch() ? cnxn : null);
rsp = new GetChildrenResponse(children);
break;
}
case OpCode.getChildren2: {
lastOp = "GETC";
GetChildren2Request getChildren2Request = new GetChildren2Request();
ByteBufferInputStream.byteBuffer2Record(request.request,
getChildren2Request);
Stat stat = new Stat();
DataNode n = zks.getZKDatabase().getNode(getChildren2Request.getPath());
if (n == null) {
throw new KeeperException.NoNodeException();
}
Long aclG;
synchronized(n) {
aclG = n.acl;
}
PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().convertLong(aclG),
ZooDefs.Perms.READ,
request.authInfo);
List<String> children = zks.getZKDatabase().getChildren(
getChildren2Request.getPath(), stat, getChildren2Request
.getWatch() ? cnxn : null);
rsp = new GetChildren2Response(children, stat);
break;
}
}
} catch (SessionMovedException e) {
// session moved is a connection level error, we need to tear
// down the connection otw ZOOKEEPER-710 might happen
// ie client on slow follower starts to renew session, fails
// before this completes, then tries the fast follower (leader)
// and is successful, however the initial renew is then
// successfully fwd/processed by the leader and as a result
// the client and leader disagree on where the client is most
// recently attached (and therefore invalid SESSION MOVED generated)
cnxn.sendCloseSession();
return;
} catch (KeeperException e) {
err = e.code();
} catch (Exception e) {
// log at error level as we are returning a marshalling
// error to the user
LOG.error("Failed to process " + request, e);
StringBuilder sb = new StringBuilder();
ByteBuffer bb = request.request;
bb.rewind();
while (bb.hasRemaining()) {
sb.append(Integer.toHexString(bb.get() & 0xff));
}
LOG.error("Dumping request buffer: 0x" + sb.toString());
err = Code.MARSHALLINGERROR;
}
long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
ReplyHeader hdr =
new ReplyHeader(request.cxid, lastZxid, err.intValue());
zks.serverStats().updateLatency(request.createTime);
cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp,
request.createTime, System.currentTimeMillis());
try {
cnxn.sendResponse(hdr, rsp, "response");
if (closeSession) {
cnxn.sendCloseSession();
}
} catch (IOException e) {
LOG.error("FIXMSG",e);
}
}
说明：对于processRequest函数，进行分段分析
if (LOG.isDebugEnabled()) {
LOG.debug("Processing request:: " + request);
}
// request.addRQRec(">final");
long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
if (request.type == OpCode.ping) { // 请求类型为PING
traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
}
if (LOG.isTraceEnabled()) {
ZooTrace.logRequest(LOG, traceMask, 'E', request, "");
}
说明：可以看到其主要作用是判断是否为PING请求，同时会根据LOG的设置确定是否进行日志记录，接着下面代码
synchronized (zks.outstandingChanges) { // 同步块
while (!zks.outstandingChanges.isEmpty()
&& zks.outstandingChanges.get(0).zxid <= request.zxid) { // outstandingChanges不为空且首个元素的zxid小于等于请求的zxid
// 移除首个元素
ChangeRecord cr = zks.outstandingChanges.remove(0);
if (cr.zxid < request.zxid) { // 若Record的zxid小于请求的zxid
LOG.warn("Zxid outstanding "
+ cr.zxid
+ " is less than current " + request.zxid);
}
if (zks.outstandingChangesForPath.get(cr.path) == cr) { // 根据路径得到Record并判断是否为cr
// 移除cr的路径对应的记录
zks.outstandingChangesForPath.remove(cr.path);
}
}
if (request.hdr != null) { // 请求头不为空
// 获取请求头
TxnHeader hdr = request.hdr;
// 获取请求事务
Record txn = request.txn;
// 处理事务
rc = zks.processTxn(hdr, txn);
}
// do not add non quorum packets to the queue.
if (Request.isQuorum(request.type)) { // 只将quorum包（事务性请求）添加进队列
zks.getZKDatabase().addCommittedProposal(request);
}
}
说明：同步块处理，当outstandingChanges不为空且其首元素的zxid小于等于请求的zxid时，就会一直从outstandingChanges中取出首元素，并且对outstandingChangesForPath做相应的操作，若请求头不为空，则处理请求。若为事务性请求，则提交到ZooKeeper内存数据库中。对于processTxn函数而言，其最终会调用DataTree的processTxn，即内存数据库结构的DataTree的处理事务函数，而判断是否为事务性请求则是通过调用isQuorum函数，会改变服务器状态的（事务性）请求就是Quorum。之后调用addCommittedProposal函数将请求添加至ZKDatabase的committedLog结构中，方便follower快速同步。
接下来会根据请求的类型进行相应的操作，如对于PING请求而言，其处理如下
case OpCode.ping: { // PING请求
// 更新延迟
zks.serverStats().updateLatency(request.createTime);
lastOp = "PING";
// 更新响应的状态
cnxn.updateStatsForResponse(request.cxid, request.zxid, lastOp,
request.createTime, System.currentTimeMillis());
// 设置响应
cnxn.sendResponse(new ReplyHeader(-2,
zks.getZKDatabase().getDataTreeLastProcessedZxid(), 0), null, "response");
return;
}
说明：其首先会根据请求的创建时间来更新Zookeeper服务器的延迟，updateLatency函数中会记录最大延迟、最小延迟、总的延迟和延迟次数。然后更新响应中的状态，如请求创建到响应该请求总共花费的时间、最后的操作类型等。然后设置响应后返回。而对于创建会话请求而言，其处理如下
case OpCode.createSession: { // 创建会话请求
// 更新延迟
zks.serverStats().updateLatency(request.createTime);
lastOp = "SESS";
// 更新响应的状态
cnxn.updateStatsForResponse(request.cxid, request.zxid, lastOp,
request.createTime, System.currentTimeMillis());
// 结束会话初始化
zks.finishSessionInit(request.cnxn, true);
return;
}
说明：其首先还是会根据请求的创建时间来更新Zookeeper服务器的延迟，然后设置最后的操作类型，然后更新响应的状态，之后调用finishSessionInit函数表示结束会话的初始化。其他请求与此类似，之后会根据其他请求再次更新服务器的延迟，设置响应的状态等，最后使用sendResponse函数将响应发送给请求方，其处理流程如下
// 获取最后处理的zxid
long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
// 响应头
ReplyHeader hdr =
new ReplyHeader(request.cxid, lastZxid, err.intValue());
// 更新服务器延迟
zks.serverStats().updateLatency(request.createTime);
// 更新状态
cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp,
request.createTime, System.currentTimeMillis()); try {
// 返回响应
cnxn.sendResponse(hdr, rsp, "response");
if (closeSession) {
// 关闭会话
cnxn.sendCloseSession();
}
} catch (IOException e) {
LOG.error("FIXMSG",e);
}
三、总结
本篇博文分析了请求处理链的FinalRequestProcessor，其通常是请求处理链的最后一个处理器，而对于请求处理链部分的分析也就到这里，还有其他的处理器再使用时再进行分析，也谢谢各位园友观看~