在zookeeper中，follower也可以接收客戶端連接，處理客戶端請求，本文將分析follower處理客戶端請求的流程：

• 讀請求處理
• 寫請求轉發與響應

follower接收轉發客戶端請求

網路層接收客戶端數據包

leader、follower都會啟動ServerCnxnFactory組件，用來接收客戶端連接、讀取客戶端數據包、將客戶端數據包轉發給zk應用層。

在"zookeeper源碼(08)請求處理及數據讀寫流程"一文中已經介紹，ServerCnxn在讀取到客戶端數據包之後，會調用zookeeperServer的processConnectRequest或processPacket方法：

• processConnectRequest方法：創建session
• processPacket方法：處理業務請求

processConnectRequest創建session

• 會使用sessionTracker生成sessionId、創建session對象
• 生成一個密碼
• 提交一個createSession類型Request並提交給業務處理器

long createSession(ServerCnxn cnxn, byte[] passwd, int timeout) {
    // 生成sessionId、創建session對象
    long sessionId = sessionTracker.createSession(timeout);
    // 生成密碼
    Random r = new Random(sessionId ^ superSecret);
    r.nextBytes(passwd);
    // 提交createSession類型Request
    CreateSessionTxn txn = new CreateSessionTxn(timeout);
    cnxn.setSessionId(sessionId);
    Request si = new Request(cnxn, sessionId, 0, OpCode.createSession, RequestRecord.fromRecord(txn), null);
    submitRequest(si);
    return sessionId;
}

processPacket處理業務請求

• 封裝Request
• 驗證largeRequest
• 提交業務層處理器

Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(), h.getType(), request, cnxn.getAuthInfo());
int length = request.limit();
if (isLargeRequest(length)) {
    // checkRequestSize will throw IOException if request is rejected
    checkRequestSizeWhenMessageReceived(length);
    si.setLargeRequestSize(length);
}
si.setOwner(ServerCnxn.me);
submitRequest(si);

FollowerRequestProcessor處理器

在follower端，客戶端請求會由FollowerRequestProcessor處理：

1. 把請求提交下游CommitProcessor處理器
2. 寫請求轉發給leader處理
3. 讀請求經過CommitProcessor直接轉發給FinalRequestProcessor處理器，直接查詢數據返回給客戶端

public void run() {
    try {
        while (!finished) {

            Request request = queuedRequests.take();

            // Screen quorum requests against ACLs first 略

            // 轉發給CommitProcessor處理器
            // 提交到queuedRequests隊列
            // 寫請求還會提交到queuedWriteRequests隊列
            maybeSendRequestToNextProcessor(request);

            // ...

            // 寫請求需要轉發給leader處理
            switch (request.type) {
            case OpCode.sync:
                zks.pendingSyncs.add(request); // 待同步命令
                zks.getFollower().request(request);
                break;
            case OpCode.create:
            case OpCode.create2:
            case OpCode.createTTL:
            case OpCode.createContainer:
            case OpCode.delete:
            case OpCode.deleteContainer:
            case OpCode.setData:
            case OpCode.reconfig:
            case OpCode.setACL:
            case OpCode.multi:
            case OpCode.check:
                zks.getFollower().request(request);
                break;
            case OpCode.createSession:
            case OpCode.closeSession:
                if (!request.isLocalSession()) {
                    zks.getFollower().request(request);
                }
                break;
            }
        }
    } catch (Exception e) {
        handleException(this.getName(), e);
    }
}

轉發leader

zks.getFollower().request(request);

Learner轉發請求：

void request(Request request) throws IOException {
    // 略

    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    DataOutputStream oa = new DataOutputStream(baos);
    oa.writeLong(request.sessionId); // sessionId
    oa.writeInt(request.cxid); // 客戶端xid
    oa.writeInt(request.type); // 業務類型
    byte[] payload = request.readRequestBytes(); // 請求體
    if (payload != null) {
        oa.write(payload);
    }
    oa.close();
    // 封裝REQUEST數據包
    QuorumPacket qp = new QuorumPacket(Leader.REQUEST, -1, baos.toByteArray(), request.authInfo);
    writePacket(qp, true); // 通過網路發給leader伺服器
}

leader處理follower請求

LearnerHandler接收REQUEST請求

case Leader.REQUEST:
    bb = ByteBuffer.wrap(qp.getData());
    sessionId = bb.getLong(); // 解析請求信息
    cxid = bb.getInt();
    type = bb.getInt();
    bb = bb.slice();
    Request si;
    if (type == OpCode.sync) {
        si = new LearnerSyncRequest(
            this, sessionId, cxid, type, RequestRecord.fromBytes(bb), qp.getAuthinfo());
    } else {
        si = new Request(null, sessionId, cxid, type, RequestRecord.fromBytes(bb), qp.getAuthinfo());
    }
    si.setOwner(this); // 用來判斷請求來自follower
    learnerMaster.submitLearnerRequest(si); // 提交給業務處理器
    requestsReceived.incrementAndGet();

submitLearnerRequest提交業務處理器：

public void submitLearnerRequest(Request si) {
    zk.submitLearnerRequest(si);
}

LeaderZooKeeperServer提交業務處理器：

public void submitLearnerRequest(Request request) {
    // 提交給PrepRequestProcessor處理器
    prepRequestProcessor.processRequest(request);
}

從此處開始走leader處理寫請求流程。

leader處理寫請求流程回顧

• PrepRequestProcessor - 做事務設置
• ProposalRequestProcessor - 發起proposal，將Request轉發給SyncRequestProcessor寫事務log、本地ack
• CommitProcessor - 讀請求直接調用下游處理器，寫請求需要等待足夠的ack之後commit再調用下游RequestProcessor處理器
• ToBeAppliedRequestProcessor - 維護toBeApplied列表
• FinalRequestProcessor - 把事務應用到ZKDatabase，提供查詢功能，返迴響應

follower處理leader數據

在follower中，Follower使用processPacket方法處理來自leader的數據包，此處看一下PROPOSAL和COMMIT的邏輯。

PROPOSAL數據包

fzk.logRequest(hdr, txn, digest);

logRequest會使用syncProcessor將事務寫入到txnlog文件，之後調用SendAckRequestProcessor處理器給leader發ack數據包。

leader收到超過半數的ack之後會發COMMIT數據包讓各個節點將事務應用到ZKDatabase中。

COMMIT數據包

fzk.commit(qp.getZxid());

CommitProcessor處理器會將其提交到committedRequests隊列，之後客戶端Request會繼續向下游FinalRequestProcessor處理器傳遞。

FinalRequestProcessor處理器

• 把事務應用到ZKDatabase中
• 提供查詢功能
• 給客戶端返迴響應