【Zookeeper】源码分析之请求处理链（四）之FinalRequestProcessor

一、前言

　　前面分析了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('') != -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，其通常是请求处理链的最后一个处理器，而对于请求处理链部分的分析也就到这里，还有其他的处理器再使用时再进行分析，也谢谢各位园友观看~