1.基本架構(gòu)

?2.ZAB協(xié)議

?? ZooKeeper并沒(méi)有完全采用Paxos算法，而是使用了一種稱為ZooKeeper Atomic Broadcast(ZAB,zookeeper原子消息廣播協(xié)議)的協(xié)議作為其數(shù)據(jù)一致性的核心算法。

??? 2.1選擇Leader需用半數(shù)通過(guò)才選舉成成功，同時(shí)集群中已經(jīng)有過(guò)半的機(jī)器與該Leader服務(wù)器完成狀態(tài)同步（數(shù)據(jù)同步）才能開(kāi)始服務(wù)。

??? 2.2所有事務(wù)請(qǐng)求必須由一個(gè)全局唯一的服務(wù)器來(lái)協(xié)調(diào)處理，這樣的服務(wù)器稱為L(zhǎng)eader服務(wù)器，而余下的其他服務(wù)器則成為Follower服務(wù)器。Leader服務(wù)器負(fù)責(zé)將一個(gè)客戶端事務(wù)請(qǐng)求轉(zhuǎn)換成一個(gè)事務(wù)Proposal(提議)，并將該P(yáng)roposal分發(fā)給集群中所有的Follower服務(wù)器。之后Leader服務(wù)器需要等待所有Follower服務(wù)器的反饋，一旦超過(guò)半數(shù)的Follower服務(wù)器進(jìn)行了正確反饋后，那么Leader就會(huì)再次向所有的Follower服務(wù)器分發(fā)Commit消息，要求其將前一個(gè)Proposal進(jìn)行提交。

3.Leader和Follower啟動(dòng)過(guò)程

4.請(qǐng)求處理

?? 4.1請(qǐng)求處理鏈

????? 4.1.1leader請(qǐng)求處理鏈

????? 4.1.2follower請(qǐng)求處理鏈

??? 4.2處理流程

??? 以creater服務(wù)端為leade為例流程如下

???? FollowerZooKeeperServer與LeaderZooKeeperServer處理流程的差別是FollowerRequestProcessor會(huì)將事務(wù)請(qǐng)求轉(zhuǎn)發(fā)給leader,SendAckRequestProcessor向leader返回事務(wù)提議正確的響應(yīng)，其他的處理鏈都是一致的。SendAckRequestProcessor和AckRequestProcessor的區(qū)別是AckRequestProcessor是leader的本地調(diào)用。FollowerRequestProcessor的事務(wù)請(qǐng)求的代碼如下

Java代碼 public?void?run()?{?????????try?{?????????????while?(!finished)?{?????????????????Request?request?=?queuedRequests.take();?????????????????if?(LOG.isTraceEnabled())?{?????????????????????ZooTrace.logRequest(LOG,?ZooTrace.CLIENT_REQUEST_TRACE_MASK,?????????????????????????????'F',?request,?"");?????????????????}?????????????????if?(request?==?Request.requestOfDeath)?{?????????????????????break;?????????????????}?????????????????//?We?want?to?queue?the?request?to?be?processed?before?we?submit?????????????????//?the?request?to?the?leader?so?that?we?are?ready?to?receive?????????????????//?the?response?????????????????nextProcessor.processRequest(request);??????????????????????????????????//?We?now?ship?the?request?to?the?leader.?As?with?all?????????????????//?other?quorum?operations,?sync?also?follows?this?code?????????????????//?path,?but?different?from?others,?we?need?to?keep?track?????????????????//?of?the?sync?operations?this?follower?has?pending,?so?we?????????????????//?add?it?to?pendingSyncs.?????????????????switch?(request.type)?{?????????????????case?OpCode.sync:?????????????????????zks.pendingSyncs.add(request);?????????????????????zks.getFollower().request(request);?????????????????????break;?????????????????case?OpCode.create:?????????????????case?OpCode.delete:?????????????????case?OpCode.setData:?????????????????case?OpCode.setACL:?????????????????case?OpCode.createSession:?????????????????case?OpCode.closeSession:?????????????????case?OpCode.multi:?????????????????????zks.getFollower().request(request);?????????????????????break;?????????????????}?????????????}?????????}?catch?(Exception?e)?{?????????????LOG.error("Unexpected?exception?causing?exit",?e);?????????}?????????LOG.info("FollowerRequestProcessor?exited?loop!");?????}??

5.數(shù)據(jù)同步

??? ZooKeeper集群數(shù)據(jù)同步分為4類，分別為直接差異化同步(DIFF)、先回滾再差異化同步(TRUNC+DIFF)、回滾同步(TRUNC)和全量同步(SNAP)。在同步之前,leader服務(wù)器先對(duì)peerLastZxid(該leader服務(wù)器最好處理的ZXID)、minCommittedLog(leader服務(wù)器提議緩存隊(duì)列committedLog中的最小ZXID)、maxCommittedLog(leader服務(wù)器提議緩存隊(duì)列committedLog中的最大ZXID)進(jìn)行初始化，然后通過(guò)這3個(gè)ZXID值進(jìn)行判斷同步類型,并進(jìn)行同步。代碼見(jiàn)LearnerHandler的run方法:

Java代碼 .....??long?peerLastZxid;??StateSummary?ss?=?null;??long?zxid?=?qp.getZxid();??long?newEpoch?=?leader.getEpochToPropose(this.getSid(),?lastAcceptedEpoch);????if?(this.getVersion()?<?0x10000)?{??????//?we?are?going?to?have?to?extrapolate?the?epoch?information??????long?epoch?=?ZxidUtils.getEpochFromZxid(zxid);??????ss?=?new?StateSummary(epoch,?zxid);??????//?fake?the?message??????leader.waitForEpochAck(this.getSid(),?ss);??}?else?{??????byte?ver[]?=?new?byte[4];??????ByteBuffer.wrap(ver).putInt(0x10000);??????QuorumPacket?newEpochPacket?=?new?QuorumPacket(Leader.LEADERINFO,?ZxidUtils.makeZxid(newEpoch,?0),?ver,?null);??????oa.writeRecord(newEpochPacket,?"packet");??????bufferedOutput.flush();??????QuorumPacket?ackEpochPacket?=?new?QuorumPacket();??????ia.readRecord(ackEpochPacket,?"packet");??????if?(ackEpochPacket.getType()?!=?Leader.ACKEPOCH)?{??????????LOG.error(ackEpochPacket.toString()??????????????????+?"?is?not?ACKEPOCH");??????????return;????????ByteBuffer?bbepoch?=?ByteBuffer.wrap(ackEpochPacket.getData());??????ss?=?new?StateSummary(bbepoch.getInt(),?ackEpochPacket.getZxid());??????leader.waitForEpochAck(this.getSid(),?ss);??}??peerLastZxid?=?ss.getLastZxid();????/*?the?default?to?send?to?the?follower?*/??int?packetToSend?=?Leader.SNAP;??long?zxidToSend?=?0;??long?leaderLastZxid?=?0;??/**?the?packets?that?the?follower?needs?to?get?updates?from?**/??long?updates?=?peerLastZxid;????/*?we?are?sending?the?diff?check?if?we?have?proposals?in?memory?to?be?able?to???*?send?a?diff?to?the???*/???ReentrantReadWriteLock?lock?=?leader.zk.getZKDatabase().getLogLock();??ReadLock?rl?=?lock.readLock();??try?{??????rl.lock();??????????????final?long?maxCommittedLog?=?leader.zk.getZKDatabase().getmaxCommittedLog();??????final?long?minCommittedLog?=?leader.zk.getZKDatabase().getminCommittedLog();??????LOG.info("Synchronizing?with?Follower?sid:?"?+?sid??????????????+"?maxCommittedLog=0x"+Long.toHexString(maxCommittedLog)??????????????+"?minCommittedLog=0x"+Long.toHexString(minCommittedLog)??????????????+"?peerLastZxid=0x"+Long.toHexString(peerLastZxid));????????LinkedList