zookeeper原理

理解zookeeper需要了解paxos算法，zookeeper是paxos算法的工业级实现。

#### Paxos算法

如果Paxos世界上只有一种消息传递的一致性算法。那就是paxos算法，其他算法都是paxos的不完整版。

paxos算法有三个角色分别是proposal、accept、learner。

1. proposal获取唯一自增序列n

2. proposal发送提案+n给accept

3. accept判读是否通过提案并给proposal并给出判读

4. 如果通过learner（learner包括accept）向proposal学习

在上述步骤中proposql有可能有多个，同时给accept发送消息会存在accept无法给出应有判读的活锁问题

### zookeeper的三个数据、四个状态、三种模式、leader选举、消息同步

zookeeper协议是paxos的实现，Zookeeper Atomic Broadcast简称ZAB支持**崩溃恢复**。

#### 三个数据

1. zxid Long包括高32位为epoch低32位为事务id在zookeeper

2. epoch逻辑时钟世纪，记录zookeeper leader的生命周期

3. xid为事务id每次提交proposal时+1

#### 四个状态

1. LOOKING 选举

2. FOLLOWING accept

3. LEADING 学习

4. LEARNING 领导

#### 三种模式

1. 恢复模式 zookeeper选举以及选举后消息和状态的同步

2. 广播模式初始化广播与更新广播

3. 同步模式初始化同步与更新同步

#### leader选举

1. 广播自己的选票推荐自己为leader

2. 接收别人选票如果比自己的事务id大则尝试计票（忽略掉Observe的选票）

3. 如果选票小于自己的事务id则丢弃

4. 如果等于则尝试计票

5. 判断是否过半以及是否有比自己存在更好的选票

6. 如果过半且都是最优票则结束投票如果没有则继续接受选票

注意：每个票都和自己进行比较谁更适合，如果发现有选票比自己更适合会清空票箱

#### 消息同步

1. 接受客户端的请求如果不是leader转到leader

2. leader生成唯一zxid+内容生成提案

3. accept判断提案zxid大于最大的zxid则向leader发送自己的zxid+机器标识

4. leader进行统计如果大于半数提交提案通知learner学习（如果超过半数accept确认收到）响应客户

#### zookeeper选举代码

```java

public Vote lookForLeader() throws InterruptedException {

// ------------------- 创建选举对象，做选举前的初始化工作 ----------------

// jmx，是Oracle提供的一种分布式应用程序监控技术

try {

self.jmxLeaderElectionBean = new LeaderElectionBean();

MBeanRegistry.getInstance().register(

self.jmxLeaderElectionBean, self.jmxLocalPeerBean);

} catch (Exception e) {

LOG.warn("Failed to register with JMX", e);

self.jmxLeaderElectionBean = null;

}

if (self.start_fle == 0) {

// 记录选举的开始时间点

self.start_fle = Time.currentElapsedTime();

}

try {

// 用于记录当前主机收到的来自于其它主机的投票信息

// 该集合相当于“票箱”，提案是否过半，就是对该集合中的选票进行统计

// key：投票者的sid

// value：选票

HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();

// 记录当前Server所投出的所有选票

HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

// 初始化本次“通知”发出后的超时时限

int notTimeout = finalizeWait;

// ------------------- 将自己作为初始化Leader投出去 ---------------

synchronized(this){

// 逻辑时钟加一

logicalclock.incrementAndGet();

// 更新提案，将自己作为Leader

updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());

}

LOG.info("New election. My id = " + self.getId() +

", proposed zxid=0x" + Long.toHexString(proposedZxid));

// 将新的选票发送出去

sendNotifications();

// ------------- 验证选票并与大家的选票进行对比，看谁更适合做Leader ------------

* Loop in which we exchange notifications until we find a leader

while ((self.getPeerState() == ServerState.LOOKING) &&

(!stop)){

* Remove next notification from queue, times out after 2 times

* the termination time

// 获取头节点的下一个节点的元素，并将头节点删除

Notification n = recvqueue.poll(notTimeout,

TimeUnit.MILLISECONDS);

* Sends more notifications if haven't received enough.

* Otherwise processes new notification.

if(n == null){

if(manager.haveDelivered()){

// 执行到这里说明当前主机与集群没有失联，则重新发送自己的选票

// 重新发送的目的是为了再次接收其它主机的选票

sendNotifications();

} else {

// 执行到这里，说明当前主机与集群失联，则重新连接

// 若当前主机与集群失联，则其它主机一定不会收齐选票，则其它主机一定会

// 重新发送它们的选票，则当前主机只需“坐等”即可

manager.connectAll();

}

* Exponential backoff

int tmpTimeOut = notTimeout*2;

notTimeout = (tmpTimeOut < maxNotificationInterval?

tmpTimeOut : maxNotificationInterval);

LOG.info("Notification time out: " + notTimeout);

}

// 从接收队列recvqueue中获取到的通知不为null

// 判断通知的发送者是否具有投票权，且通知的推荐者是否具有被选举权

else if(validVoter(n.sid) && validVoter(n.leader)) {

* Only proceed if the vote comes from a replica in the

* voting view for a replica in the voting view.

switch (n.state) {

case LOOKING:

// If notification > current, replace and send messages out

// 判断通知所处选举的epoch与当前主机选举的epoch的大小关系

// 若n的epoch较当前主机的逻辑时钟大，则说明当前主机的选举工作已经过时，

// 已经与n所发出的选举不在同一轮，所以当前主机需要更新自己的选举工作了

if (n.electionEpoch > logicalclock.get()) {

// 更新逻辑时钟：将n的逻辑时钟取代我们自己的逻辑时钟

// 经过该语句后，当前主机的逻辑时钟与n的逻辑时钟就相同了

logicalclock.set(n.electionEpoch);

// 清空“票箱”

recvset.clear();

// 判断n与当前主机谁更适合做Leader

if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,

getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {

// 若n更适合做Leader，则当前主机将提案更新为n

updateProposal(n.leader, n.zxid, n.peerEpoch);

} else {

// 若当前主机的提案更适合，按理说应该不用再更新提案了，为什么这里又更新了？

// 因为前面我们已经更新过了逻辑时钟了，将逻辑时钟已经更新为了n的epoch了

updateProposal(getInitId(),

getInitLastLoggedZxid(),

getPeerEpoch());

}

// 将更新过的提案发布出去

sendNotifications();

// n的epech较当前主机的逻辑时钟小，说明n的选举已经过时，则n的选票对于

// 当前主机已经没有任何统计的意义，所以直接结束switch-case，重新再获取

// 一个notification进行判断

} else if (n.electionEpoch < logicalclock.get()) {

if(LOG.isDebugEnabled()){

LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"

+ Long.toHexString(n.electionEpoch)

+ ", logicalclock=0x" + Long.toHexString(logicalclock.get()));

}

// 跳出switch-case

break;

// n.electionEpoch = logicalclock.get() 的情况

// 判断n与当前主机的提案，谁更适合当Leader

// 若n的提案更适合，则返回true，否则返回false

} else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,

proposedLeader, proposedZxid, proposedEpoch)) {

// 更新提案

updateProposal(n.leader, n.zxid, n.peerEpoch);

// 将更新过的提案发布出去

sendNotifications();

}

if(LOG.isDebugEnabled()){

LOG.debug("Adding vote: from=" + n.sid +

", proposed leader=" + n.leader +

", proposed zxid=0x" + Long.toHexString(n.zxid) +

", proposed election epoch=0x" + Long.toHexString(n.electionEpoch));

}

// 将有效选票放到到“票箱”

recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

// 判断当前选举是否终止

// 判断当前主机提案形成的选票在recvset集合（票箱）中是否过半

if (termPredicate(recvset,

new Vote(proposedLeader, proposedZxid,

logicalclock.get(), proposedEpoch))) {

// Verify if there is any change in the proposed leader

// 判断剩余的通知中是否存在“比当前过半选票的通知更适合的”通知

while((n = recvqueue.poll(finalizeWait,

TimeUnit.MILLISECONDS)) != null){

// 判断n与当前的提案谁更适合做Leader，但需要注意，当前的

// 提案支持者已经过半了

if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,

proposedLeader, proposedZxid, proposedEpoch)){

// 将n重新塞回到recvqueue中

recvqueue.put(n);

break;

}

// 若前面的while是从第一个出口跳出的(while的判断条件)，则说明

// 剩余的通知中不存在“比当前过半选票的通知更适合的”通知

// 若前面的while是从第二个出口跳出的(break)，则说明

// 剩余的通知中已经找到“比当前过半选票的通知更适合的”通知

* This predicate is true once we don't read any new

* relevant message from the reception queue

// 若n的值为null，则说明while是从第一个出口出来的，则说明

// 剩余的通知中不存在“比当前过半选票的通知更适合的”通知

if (n == null) {

// 修改主机状态

self.setPeerState((proposedLeader == self.getId()) ?

ServerState.LEADING: learningState());

// 形成最终选票，并返回，以备其它使用

Vote endVote = new Vote(proposedLeader,

proposedZxid,

logicalclock.get(),

proposedEpoch);

leaveInstance(endVote);

return endVote;

}

break;

case OBSERVING:

LOG.debug("Notification from observer: " + n.sid);

break;

case FOLLOWING:

case LEADING:

* Consider all notifications from the same epoch

* together.

if(n.electionEpoch == logicalclock.get()){

recvset.put(n.sid, new Vote(n.leader,

n.zxid,

n.electionEpoch,

n.peerEpoch));

if(ooePredicate(recvset, outofelection, n)) {

self.setPeerState((n.leader == self.getId()) ?

ServerState.LEADING: learningState());

Vote endVote = new Vote(n.leader,

n.zxid,

n.electionEpoch,

n.peerEpoch);

leaveInstance(endVote);

return endVote;

}

* Before joining an established ensemble, verify

* a majority is following the same leader.

outofelection.put(n.sid, new Vote(n.version,

n.leader,

n.zxid,

n.electionEpoch,

n.peerEpoch,

n.state));

if(ooePredicate(outofelection, outofelection, n)) {

synchronized(this){

logicalclock.set(n.electionEpoch);

self.setPeerState((n.leader == self.getId()) ?

ServerState.LEADING: learningState());

}

Vote endVote = new Vote(n.leader,

n.zxid,

n.electionEpoch,

n.peerEpoch);

leaveInstance(endVote);

return endVote;

}

break;

default:

LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)",

n.state, n.sid);

break;

}

} else {

if (!validVoter(n.leader)) {

LOG.warn("Ignoring notification for non-cluster member sid {} from sid {}", n.leader, n.sid);

}

if (!validVoter(n.sid)) {

LOG.warn("Ignoring notification for sid {} from non-quorum member sid {}", n.leader, n.sid);

}

return null;

} finally {

try {

if(self.jmxLeaderElectionBean != null){

MBeanRegistry.getInstance().unregister(

self.jmxLeaderElectionBean);

}

} catch (Exception e) {

LOG.warn("Failed to unregister with JMX", e);

}

self.jmxLeaderElectionBean = null;

LOG.debug("Number of connection processing threads: {}",

manager.getConnectionThreadCount());

}

```