zookeeper原理之Leader選舉原始碼分析
阿新 • • 發佈：2020-08-18
Zookeeper 的一致性
Zookeeper 的來源
對於 zookeeper 的一致性問題，有很多朋友有疑問，我這邊再幫大家從來源層面梳理一遍一致性的問題。上篇，我們講到了 zookeeper 的來源，是來自於 google chubby。為了解決在分散式環境下，如何從多個 server 中選舉出 master server。那麼這多個 server 就需要涉及到一致性問題，這個一致性體現的是多個server 就 master 這個投票在分散式環境下達成一致性。簡單來說就是最終聽誰的。但是在網路環境中由於網路的不可靠性，會存在訊息丟失和或者被篡改等問題。所以如何在這樣一個環境中快速並且正確的在多個server 中對某一個數據達成一致性並且保證不論發生任何異常，都不會破壞整個系統一致性呢？所以在 Lamport 大神設計了一套 Paxos 的演算法，多個 server 基於這個演算法就可以達成一致。而 google chubby 就是基於 paxos 演算法的實現，用來實現分散式鎖服務。並且提供了 master 選舉的服務。
Paxos 在 Chubby 中的應用
- LOOKING，競選狀態。
- FOLLOWING，隨從狀態，同步 leader 狀態，參與投票。
- OBSERVING，觀察狀態,同步 leader 狀態，不參與投票。LEADING，領導者狀態。
- LEADING，領導者狀態。
- 優先比較 epoch
- 其次檢查 ZXID。ZXID 比較大的伺服器優先作為 Leader
- 如果 ZXID 相同，那麼就比較 myid。myid 較大的伺服器作為Leader 伺服器。
protected void initializeAndRun(String[] args) throws ConfigException, IOException{
//這段程式碼比較簡單，設定配置引數，如果 args 不為空，可以基於外部的配置路徑來進行解析
QuorumPeerConfig config = new QuorumPeerConfig();
if (args.length == 1) {
config.parse(args[0]);
}
// 這裡啟動了一個執行緒，來定時對日誌進行清理，從命名來看也很容易理解
DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config.getDataDir(),
config.getDataLogDir(), config.getSnapRetainCount(), config.getPurgeInterval());
purgeMgr.start();
//如果是叢集模式，會呼叫 runFromConfig.servers 實際就是我們在 zoo.cfg 裡面配置的叢集節點
if (a config.servers.size() > 0) {
runFromConfig(config);
} else {//否則直接執行單機模式
LOG.warn("Either no config or no quorum defined in config, running " + " in standalone mode");
// there is only server in the quorum -- run as standalone
ZooKeeperServerMain.main(args);
}
}
runFromConfig
從名字可以看出來，是基於配置檔案來進行啟動。
所以整個方法都是對引數進行解析和設定 ，因為這些引數暫時還沒用到，所以沒必要去看。直接看核心的程式碼quorumPeer.start()， 啟動一個執行緒，那麼從這句程式碼可以看出來QuorumPeer 實際是繼承了執行緒。那麼它裡面一定有一個 run 方法：
public void runFromConfig(QuorumPeerConfig config) throws IOException {
try {
ManagedUtil.registerLog4jMBeans();
} catch (JMException e) {
LOG.warn("Unable to register log4j JMX control", e);
}
LOG.info("Starting quorum peer");
try {
ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns());
quorumPeer = getQuorumPeer();
quorumPeer.setQuorumPeers(config.getServers());
quorumPeer.setTxnFactory(new FileTxnSnapLog( new File(config.getDataLogDir()), new File(config.getDataDir())));
quorumPeer.setElectionType(config.getElectionAlg());
quorumPeer.setMyid(config.getServerId());
quorumPeer.setTickTime(config.getTickTime());
quorumPeer.setInitLimit(config.getInitLimit());
quorumPeer.setSyncLimit(config.getSyncLimit());
//投票決定方式，預設超過半數就通過
quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());
quorumPeer.setCnxnFactory(cnxnFactory);
quorumPeer.setQuorumVerifier(config.getQuorumVerifi er());
quorumPeer.setClientPortAddress(config.getClientPor tAddress());
quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
quorumPeer.setLearnerType(config.getPeerType());
quorumPeer.setSyncEnabled(config.getSyncEnabled());
// sets quorum sasl authentication
configurationsquorumPeer.setQuorumSaslEnabled(config.quorumEnableSasl);
if(quorumPeer.isQuorumSaslAuthEnabled()){
quorumPeer.setQuorumServerSaslRequired(config.quorumServerRequireSasl);
quorumPeer.setQuorumLearnerSaslRequired(config.quorumLearnerRequireSasl);
quorumPeer.setQuorumServicePrincipal(config.quorumServicePrincipal);
quorumPeer.setQuorumServerLoginContext(config.quorumServerLoginContext);
quorumPeer.setQuorumLearnerLoginContext(config.quorumLearnerLoginContext);
}
quorumPeer.setQuorumCnxnThreadsSize(config.quorumCnxnThreadsSize);
quorumPeer.initialize();
//啟動主執行緒
quorumPeer.start();
quorumPeer.join();
} catch (InterruptedException e) {
// warn, but generally this is ok
LOG.warn("Quorum Peer interrupted", e);
}
}
QuorumPeer.start
QuorumPeer.start 方法，重寫了 Thread 的 start。也就是線上程啟動之前，會做以下操作
1. 通過 loadDataBase 恢復快照資料
2. cnxnFactory.start() 啟動 zkServer，相當於使用者可以通過 2181 這個埠進行通訊了，這塊後續在講。我們還是以 leader 選舉為主線
@Override
public synchronized void start() {
loadDataBase();
cnxnFactory.start();
startLeaderElection();
super.start();
}
startLeaderElection
看到這個方法，有沒有兩眼放光的感覺？沒錯，前面鋪墊了這麼長，終於進入 leader 選舉的方法了
synchronized public void startLeaderElection() {
try {
// 構建一個票據，用於投票
currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
} catch (IOException e) {
RuntimeException re = new RuntimeException(e.getMessage());
re.setStackTrace(e.getStackTrace());
throw re;
}
// 這個 getView 返回的就是在配置檔案中配置的server.myid=ip:port:port。view 在哪裡解析的呢？
for (QuorumServer p : getView().values()) {
if (p.id == myid) {// 獲得當前 zkserver myid 對應的 ip 地址
myQuorumAddr = p.addr;
break;
}
}
if (myQuorumAddr == null) {
throw new RuntimeException("My id " + myid + " not in the peer list");
} // 根據 electionType 匹配對應的選舉演算法,electionType 預設值為 3.可以在配置檔案中動態配置
if (electionType == 0) {
try {
udpSocket = new DatagramSocket(myQuorumAddr.getPort());
responder = new ResponderThread();
responder.start();
} catch (SocketException e) {
throw new RuntimeException(e);
}
}
this.electionAlg = createElectionAlgorithm(electionType);
}
quorumPeer. createElectionAlgorithm
根據對應的標識建立選舉演算法
protected Election createElectionAlgorithm(int electionAlgorithm) {
Election le = null;
// TODO: use a factory rather than a switch
switch (electionAlgorithm) {
case 0:
le = new LeaderElection(this);
break;
case 1:
le = new AuthFastLeaderElection(this);
break;
case 2:
le = new AuthFastLeaderElection(this, true);
break;
case 3:
qcm = createCnxnManager();
QuorumCnxManager.Listener listener = qcm.listener;
if (listener != null) {
listener.start(); // 啟動監聽器，這個監聽具體做什麼的暫時不管，後面遇到需要了解的地方再回過頭來看
le = new FastLeaderElection(this, qcm);// 初始化 FastLeaderElection
} else {
LOG.error("Null listener when initializing cnx manager");
}
break;
default:
assert false;
}
return le;
}
FastLeaderElection
初始化FastLeaderElection，QuorumCnxManager 是一個很核心的物件，用來實現領導選舉中的網路連線管理功能，這個後面會用到
public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
this.stop = false;
this.manager = manager;
starter(self, manager);
}
FastLeaderElection. starter
starter 方法裡面，設定了一些成員屬性，並且構建了兩個阻塞佇列，分別是 sendQueue 和 recvqueue。並且例項化了一個 Messager
private void starter(QuorumPeer self, QuorumCnxManager manager) {
this.self = self;
proposedLeader = -1;
proposedZxid = -1;
sendqueue = new LinkedBlockingQueue<ToSend>();
recvqueue = new LinkedBlockingQueue<Notification>();
this.messenger = new Messenger(manager);
}
Messenger
在 Messenger 裡面構建了兩個執行緒，一個是 WorkerSender，一個是WorkerReceiver。 這兩個執行緒是分別用來發送和接收訊息的執行緒。具體做什麼，暫時先不分析。
Messenger(QuorumCnxManager manager) {
this.ws = new WorkerSender(manager);
Thread t = new Thread(this.ws,
"WorkerSender[myid=" + self.getId() + "]");
t.setDaemon(true);
t.start();
this.wr = new WorkerReceiver(manager);
t = new Thread(this.wr,
"WorkerReceiver[myid=" + self.getId() + "]");
t.setDaemon(true);
t.start();
}
階段性總結
ok，分析到這裡，先做一個簡單的總結，通過一個流程圖把前面部分的功能串聯起來。
至此，先做如上總結~