2. 程式人生 > >ZooKeeper原始碼分析：Quorum請求的整個流程

ZooKeeper原始碼分析：Quorum請求的整個流程

阿新 • • 發佈：2019-01-28

Quorum請求是轉發給Leader處理，並且需要得一個Follower Quorum確認的請求。這些請求包括：

1）znode的寫操作(OpCode.create，OpCode.delete，OpCode.setData，OpCode.setACL)

2）Session的建立和關閉操作（OpCode.createSession和OpCode.closeSession)

3）OpCode.multi操作。

本博文分析了Client, Follower和Leader協同完成Quorum請求的過程。另外需注意的是OpCode.sync請求也需要轉發給Leader, 但不需要得到一個Follower Quorum確認。本文也會提到OpCode.sync操作。

資料結構

Request型別物件：Server內部傳遞的資料結構。

屬性 說明
sessionId  會話ID
cxid 客戶端事務ID
type 操作型別, 如OpCode.setData
request 請求Record物件，如SetDataRequest
cnxn Server和Client端的連線物件
hdr 請求事務頭TxnHeader
txn 請求事務體Record，如OpCode.setData請求，則是SetDataTxn型別物件
zxid ZooKeeper事務ID
authInfo 認證資訊
createTime 建立時間
owner 所有者
e 處理過程中的異常

QuorumPacket型別物件：用於ZooKeeper伺服器之間傳遞的資料包。

屬性 說明
zxid ZooKeeper事務ID
data 資料包的資料:
在Leader.REQUEST中，資料依次如下：
Request.sessionId
Request.cxid
Request.type
Request.request
在Leader.PROPOSAL中，資料依次如下：
Request.hdr
Request.txn
在Leader.ACK中，為null
在Leader.COMMIT中，為null
authinfo 認證資訊

Quorum請求流程

假設拓撲結構如下圖，Client A和Follower A建立連線。

Image(8)

資料流程圖如下。在圖中，連線線說明前的數字表示事件發的生時序，主時序是直接使用一個數字表示，並且數字越小表示越早發生（如1 Client Request是在2 Request之前發生）。對於和主時序併發的操作使用主時序序號後加上一個括號括起來的數字表示，如7(1)-n Request指和7 Request是併發的。7(1)-n中n表示以7(1)開頭的操作時序。

zookeeper請求-寫 - 透明背景 - 沒有註釋

我們從資料流程圖中Step 1講起：Client A 發起一個Quorum請求給Follower A。

【Client A, Step 1】Client A呼叫Quorum請求對應的方法：

如呼叫Zookeeper的建構函式，會發起OpCode.createSession請求，

如呼叫Zookeeper.setData方法，會發起OpCode.setData操作。

最終會呼叫ClientCnxn.submitRequest方法將請求放入outgoingQueue佇列中，並阻塞等待Follower A反饋。而ClientCnxn.SendThread執行緒會從outgoingQueue中取出請求，併發送給Follower A。

下面程式碼Zookeeper.setData方法: Client A構建物件傳送給Follower A

publicStat setData(finalString path,bytedata[],intversion) throwsKeeperException, InterruptedException     { finalString clientPath = path;         PathUtils.validatePath(clientPath);         //通過傳入的path構造完整serverPath finalString serverPath = prependChroot(clientPath);         //構造一個Request頭         RequestHeader h=newRequestHeader();         //設定型別為setData h.setType(ZooDefs.OpCode.setData);         //構造一個SetData請求體         SetDataRequest request =newSetDataRequest();         //設定需要修改node的serverPath         request.setPath(serverPath);         //設定需要修改的node的data         request.setData(data);         //設定需要修改的node的version         request.setVersion(version);         //構建SetDataResponse物件         SetDataResponse response =newSetDataResponse();         //提交請求，並等待返回結果         ReplyHeader r =cnxn.submitRequest(h, request, response,null);         //如果r.getErr()不能0，則表示有錯誤，丟擲異常 if(r.getErr() != 0) { throwKeeperException.create(KeeperException.Code.get(r.getErr()),                     clientPath);         } returnresponse.getStat();     }

【Follower A, Step 2,3】Follower A的NIOServerCnxn類接到了Client A的請求，會呼叫ZookeeperServer.processPacket方法。該方法會構建一個Request物件，並呼叫第一個處理器FollowerRequestProcessor的processRequest方法。該方法將Request物件放入FollowerRequestProcessor.queuedRequests佇列中。FollowerRequestProcessor處理器執行緒會迴圈從FollowerRequestProcessor.queuedRequests佇列中取出Request物件，並繼續下面步驟：

1）呼叫下一個處理器CommitProcessor的processRequest方法。該方法將Request物件放入CommitProcessor.queuedRequests佇列中；

2）通過Request.type判斷Request型別。若發現是一個Quorum請求，會直接呼叫Learner.request(request)方法。該方法將Request物件封裝成一個Leader.Request的Quorum資料包，併發送給Leader。

OpCode.sync操作也將呼叫Learner.request方法將請求轉發給Leader，但在這之前會先將Request物件加入到pendingSyncs佇列中。

FollowerRequestProcessor的run方法如下:

publicvoidrun() { try{ while(!finished) { //從queuedRequests佇列中取出Request物件 Requestrequest =queuedRequests.take(); if(LOG.isTraceEnabled()) { ZooTrace. logRequest(LOG,ZooTrace.CLIENT_REQUEST_TRACE_MASK , 'F', request,"");                 } //當request是Request.requestOfDeath，一個poison pill, 就退出while迴圈，                 //並結束FollowerRequestProcessor執行緒 if(request ==Request.requestOfDeath) { break;                 } //我們在提交這個request到leader之前，把這個request傳遞到下一個處理器。 //這樣我們就準備好從Leader那得到Response nextProcessor.processRequest(request); //只有Quorum操作和sync操作才會呼叫Follower.request方法, 轉發Leader.REQUEST資料包給Leader //sync操作和Quorum操作有一些不同， //我們需要保持跟蹤這個sync操作對於的Follower已經掛起，所有我們將它加入pendingSyncs佇列中。 switch(request.type ) { caseOpCode.sync: //將OpCode.sync放入pendingSyncs佇列中 zks.pendingSyncs .add(request); zks.getFollower().request(request); break; caseOpCode.create: caseOpCode.delete: caseOpCode.setData: caseOpCode.setACL: caseOpCode.createSession: caseOpCode.closeSession: caseOpCode.multi:                     //Quorum請求，直接呼叫Folloer.request方法 zks.getFollower().request(request); break;                 }             }         }catch(Exception e) { LOG.error("Unexpected exception causing exit", e);         } LOG.info("FollowerRequestProcessor exited loop!");     }

【Leader A, Step 4】Leader A的LearnerHandler執行緒會迴圈讀取從Learner獲得的Quorum資料包。如果資料包是Learner.REQUEST型別，則會解析Quorum資料包的內容，檢查操作型別。
如果操作型別不是OpCode.sync, 則會構造Request物件。並呼叫ZooKeeperServer.submitRequest方法(和上面Follower接收到請求所使用的submitRequest方法是同一個方法），並最終會呼叫第一個處理器PrepRequestProcessor的submitRequest方法，將Request物件放入PrepRequestProcessor.submittedRequests佇列中。

如果操作型別是OpCode.sync, 會構造Request型別的子類LearnerSyncRequest物件，並同樣呼叫PrepRequestProcessor的submitRequest方法。

LearnerHandler.run方法中對Leader.REQUEST資料包的處理程式碼如下:

publicvoidrun () { ...... caseLeader.REQUEST:            bb =ByteBuffer. wrap(qp.getData()); //從QuorumPacket中讀取sesssionId sessionId= bb.getLong(); //從QuorumPacket中讀取cxid cxid= bb.getInt(); //從QuorumPacket中讀取操作型別 type= bb.getInt();            bb = bb.slice(); Requestsi; //如果操作Code的型別是OpCode.sync,則構造LearnerSyncRequest物件 if(type==OpCode.sync){                si =newLearnerSyncRequest(this,sessionId,cxid,type, bb,qp.getAuthinfo());            } //如果操作Code的型別不是OpCode.sync, 則構造Request物件 else{                si =newRequest(null,sessionId,cxid,type, bb,qp.getAuthinfo());            } //設定owner            si.setOwner(this); //提交請求 leader.zk .submitRequest(si); break; ......    }

PrepRequestProcessor處理器執行緒會從PrepRequestProcessor.submittedRequests佇列中取出Request物件，並根據Request型別構建TxnHeader和Record物件，然後分別賦給Request.hdr和Request.txn。之後會呼叫下一個處理器ProposalRequestProcessor的processRequest方法，將Request物件傳遞給處理器ProposalRequestProcessor。（如果發現有異常會則會建立一個錯誤Record型別物件）

PrepRequestProcessor的run方法如下:

publicvoidrun(){ try{ while(true) { //從submittedRequests佇列中取去第一個request物件 Requestrequest =submittedRequests.take(); longtraceMask =ZooTrace.CLIENT_REQUEST_TRACE_MASK; //如果是OpCode.ping操作，則將traceMask設定成ZooTrace. CLIENT_PING_TRACE_MASK if(request.type ==OpCode.ping) {                     traceMask =ZooTrace. CLIENT_PING_TRACE_MASK;                 } if(LOG.isTraceEnabled()) { ZooTrace. logRequest(LOG, traceMask,'P', request,"");                 } //如果request是一個requestOfDeath, 則退出while迴圈。 if(Request.requestOfDeath == request) { break;                 } //處理請求                 pRequest(request);             }         }catch(InterruptedException e) { LOG.error("Unexpected interruption", e);         }catch(RequestProcessorExceptione) { if(e.getCause()instanceofXidRolloverException) { LOG.info(e.getCause().getMessage());             } LOG.error("Unexpected exception", e);         }catch(Exception e) { LOG.error("Unexpected exception", e);         } LOG.info("PrepRequestProcessor exited loop!");     }

PrepRequestProcessor的pRequest2Txn方法，該方法會在pRequest方法中呼叫，構建TxnHeader和Record物件。下面是關於OpCode.setData請求的程式碼:

protectedvoidpRequest2Txn(inttype,longzxid,Requestrequest,Recordrecord,booleandeserialize) throwsKeeperException,IOException,RequestProcessorException     {         request.hdr =newTxnHeader(request.sessionId , request.cxid, zxid, zks.getTime(), type); switch(type) { ..... caseOpCode.setData: //檢查session zks.sessionTracker .checkSession(request.sessionId, request.getOwner()); //將record轉成SetDataRequest型別 SetDataRequestsetDataRequest = (SetDataRequest)record; if(deserialize) //將Request.reques資料反序列化成setDataRequest物件 ByteBufferInputStream.byteBuffer2Record(request. request, setDataRequest); //獲取需要需要修改的znode的path path= setDataRequest.getPath(); //獲取記憶體資料中獲取path對於的znode資訊 nodeRecord= getRecordForPath(path); //檢查對znode是否有寫許可權                 checkACL(zks,nodeRecord.acl ,ZooDefs.Perms.WRITE,                         request.authInfo); //獲取客戶端設定的版本號 version= setDataRequest.getVersion(); //獲取節點當前版本號 intcurrentVersion =nodeRecord.stat.getVersion(); //如果客戶端設定的版本號不是-1，且不等於當前版本號，則丟擲KeeperException.BadVersionException異常 if(version!= -1 &&version!= currentVersion) { thrownewKeeperException.BadVersionException(path);                 } //version等於當前版本加1 version= currentVersion + 1; //構建SetDataTxn物件，並賦給request.txn                 request. txn =newSetDataTxn(path, setDataRequest.getData(),version); //拷貝nodeRecord nodeRecord=nodeRecord.duplicate(request.hdr.getZxid()); //將nodeRecord的當前版本號設定為version nodeRecord.stat.setVersion(version); //將nodeRecord放入outstandingChanges                 //path和nodeRecord map放入outstandingChangesForPath                 addChangeRecord(nodeRecord); break;            ......         }     }

【Leader A, Step 5,6】處理器ProposalRequestProcessor會先判斷Request物件是否是LearnerSyncRequest型別。

如果不是LearnerSyncRequest型別（也就是Quorum請求），會按如下步驟執行：

1）呼叫下一個處理器CommitProcessor的processRequest方法，將Request物件放入CommitProcessor.queuedRequests佇列中;

2）將proposal傳送到所有的Follower;

3）呼叫SyncRequestProcessor處理器的processRequest方法。該方法會將請求放入SyncRequestProcessor.queuedRequests佇列中。（【Leader A, Step 7(1)】SyncRequestProcessor執行緒會記錄Log, 然後傳遞給SendAckRequestProcessor。SendAckRequestProcessor會發送一個Leader.ACK的Quorum資料包給自己）

如果是LearnerSyncRequest型別，說明該請求是OpCode.sync操作，則會直接呼叫Leader.processSync方法。

ProposalRequestProcessor的processRequest方法如下:

publicvoidprocessRequest(Requestrequest)throwsRequestPrzocessorException{ //如果是sync操作，則呼叫Leader.processSync方法 if(requestinstanceofLearnerSyncRequest){ zks.getLeader().processSync((LearnerSyncRequest)request);         } //如果不是sync操作 else{ //傳遞到下一個處理器 nextProcessor.processRequest(request); if(request.hdr !=null) { // We need to sync and get consensus on any transactions try{ //傳送proposal給所有的follower zks.getLeader().propose(request);                 }catch(XidRolloverExceptione) { thrownewRequestProcessorException(e.getMessage(), e);                 } //呼叫SyncRequestProcessor處理器的processRequest方法

相關推薦

ZooKeeper原始碼分析：Quorum請求的整個流程

Quorum請求是轉發給Leader處理，並且需要得一個Follower Quorum確認的請求。這些請求包括： 1）znode的寫操作(OpCode.create，OpCode.delete，OpCode.setData，OpCode.setACL) 2）Sessi

Zookeeper原始碼分析：Watcher機制

1. 設定Watcher 使用Watcher需要先實現Watcher介面，並將實現類物件傳遞到指定方法中，如getChildren, exist等。Zookeeper允許在構造Zookeeper物件時候指定一個預設Watcher物件.getChildren和exit方法可以

Cassandra原始碼分析：資料寫入流程

org.apache.cassandra.thrift.CassandraServer類的add方法將接受客戶端的請求，該函式定義如下： public void add(ByteBuffer key, ColumnParent column_parent, Counter

x265原始碼分析：sao.cpp 自適應樣點補償

/* 對|num / den|四捨五入，然後前面新增符號 */ inline int32_t roundIBDI(int32_t num, int32_t den) { return num >= 0 ? ((num * 2 + den)/(den * 2)) : -((-

x265原始碼分析：main函式及CLIOptions結構體解釋

/** * 返回碼資訊： * 0 – 編碼成功； * 1 – 命令列解釋失敗； * 2 – 編碼器開啟失敗； * 3 – 生成流頭部失敗； * 4 – 編碼出錯； * 5 – 開啟csv檔案失敗. */ int main(int argc, char **argv) {

Dubbo原始碼分析：Dubbo自己實現的IOC

  在建立自適應例項時，都會呼叫ExtensionLoader的injectExtension方法： @SuppressWarnings("unchecked") private T createAdaptiveExtension() { try {

Spring Developer Tools 原始碼分析：三、重啟自動配置'

接上文 Spring Developer Tools 原始碼分析：二、類路徑監控，接下來看看前面提到的這些類是如何配置，如何啟動的。 spring-boot-devtools 使用了 Spring Boot 的自動配置方式，我們先關注本地開發環境中自動重啟的部分。 在 LocalDevToolsAut

JAVA常用集合原始碼分析：HashSet

序言 在上一篇文章中，我們介紹了HashMap，其實本來想自己完成原始碼分析的一系列文章的，但是HashMap的原始碼著實是複雜，看的我腦殼疼。。於是就自己去找了找大牛們的文章反覆看，後面總算有了點門道了，大致知道了HashMap的原理，然後轉載了一篇我認為總結的比較好的文章到我的部落格裡，供大

JAVA常用集合原始碼分析：HashMap

我們這篇文章就來試著分析下 HashMap 的原始碼，由於 HashMap 底層涉及到太多方面，一篇文章總是不能面面俱到，所以我們可以帶著面試官常問的幾個問題去看原始碼： 瞭解底層如何儲存資料的 HashMap 的幾個主要方法 HashMap 是如何確定元素儲存位置的以及如何處

JAVA常用集合原始碼分析：LinkedList

概述 上一篇我們介紹了ArrayList,我們知道它的底層是基於陣列實現的，提到陣列，我們就馬上會想到它的兄弟連結串列，今天我們要介紹的LinkedList就是基於連結串列實現的。 繼承結構 public class LinkedList<E> extends Abs

JAVA常用集合原始碼分析：ArrayList

ArrayList簡介 ArrayList 是一個動態陣列，所謂動態，是相對陣列來說的，我們知道當我們在使用陣列的時候必須指定大小，而且大小隻能是固定的，有時候就很不方便，讓人不爽。而我們的ArrayList恰恰解決了這一痛點，讓我們可以不受束縛地使用陣列。 閱讀方法 看繼承結構與

QT原始碼分析：QObject

　　QT框架裡面最大的特色就是在C++的基礎上增加了元物件系統（Meta-Object System），而元物件系統裡面最重要的內容就是訊號與槽機制，這個機制是在C++語法的基礎上實現的，使用了函式、函式指標、回撥函式等概念。當然與我們自己去寫函式所不同的是槽與訊號機制會自動幫我們生成部分程式碼，比如我們寫的

Tomcat原始碼分析：一、tomcat元件認識

    前言      最近一次上線過程中出現了jekin是自動化部署的web環境跟本地開發環境不一致的情況，導致生產環境應用訪問失敗，因此閱讀tomcat原始碼，以加深對web的認識。 基本元件 在閱讀原始碼之前，最好是對整個應

J.U.C原始碼分析：CountDownLatch

        CountDownLoad是在併發程式設計中使用較多的一個類，可以完成多個執行緒之間的相互等待和協作，原始碼內容不多但功能強大且使用場景複雜多樣。         原始碼

SIFT原理與原始碼分析：DoG尺度空間構造

《SIFT原理與原始碼分析》系列文章索引：http://blog.csdn.net/xiaowei_cqu/article/details/8069548 尺度空間理論   自然界中的物體隨著觀測尺度不同有不同的表現形態。例如我們形容建築物用“米”，觀測分子、原子等用“納米”。

Zookeeper 原始碼（七）請求處理

Zookeeper 原始碼（七）請求處理 以單機啟動為例講解 Zookeeper 是如何處理請求的。先回顧一下單機時的請求處理鏈。 // 單機包含 3 個請求鏈：PrepRequestProcessor -> SyncRequestProcessor -> FinalRequestProces

muduo原始碼分析：TcpConnection類

前言 前面學習了TcpServer的實現，TcpServer對每個連線都會新建一個TcpConnection（使用shared_ptr管理）。接下來學習一下TcpConnection的設計細節。   連線狀態 muduo對於一個連線的從生到死進行了狀態的定義，類似一個狀態機

muduo原始碼分析：TcpServer類

上篇博文學習了Acceptor class 的實現，它僅僅是對Channel和Socket的簡單封裝，對使用者來說簡單易用。這得益於底層架構Reactor。接下來，開始學習muduo對於建立連線的處理。這屬於muduo提到的三個半事件中的第一個。可以想一下，TcpServer class應該也是對A

muduo原始碼分析：Acceptor類

Acceptor用於接受（accept）客戶端的連線，通過設定回撥函式通知使用者。它只在muduo網路庫內部的TcpServer使用，由TcpServer控制它的生命期。 實際上，Acceptor只是對 Channel 的封裝，通過Channel關注listenfd的&nbs

muduo原始碼分析：ThreadPool 執行緒池的實現

原始碼: https://github.com/chenshuo/muduo/blob/master/muduo/base/ThreadPool.h https://github.com/chenshuo/muduo/blob/master/muduo/base/ThreadPool.cc