Leader 选举与多节点同步

Leader 的协调过程

底层实现

广播模式

public class Leader(){

 HashSet<LearnerHandler> forwardingFollowers;

 public List<LearnerHandler> getForwardingFollowers() {

   synchronized (forwardingFollowers) {

       return new ArrayList<LearnerHandler>(forwardingFollowers);

 }

}

public boolean isQuorumSynced(QuorumVerifier qv) {

  synchronized (forwardingFollowers) {

   for (LearnerHandler learnerHandler: forwardingFollowers){

       if(learnerHandler.synced()){

         ids.add(learnerHandler.getSid());

       }

   }

  }

}

request.setTxn(new SetDataTxn(ZooDefs.CONFIG_NODE, request.qv.toString().getBytes(), -1));

恢复模式

• 恢复模式：
当 ZooKeeper 集群中的 Leader 服务器崩溃后，ZooKeeper 集群就采用恢复模式的方式进行工作，在这个工程中，ZooKeeper 集群会首先进行 Leader 节点服务器的重新选择，之后在选举出 Leader 服务器后对系统中所有的服务器进行数据同步进而保证集群中服务器上的数据的一致性。

• 广播模式：
当 ZooKeeper 集群中具有 Leader 服务器，并且可以正常工作时，集群中又有新的 Follower 服务器加入 ZooKeeper 中参与工作，这种情况常常发生在系统性能到达瓶颈，进而对系统进行动态扩容的使用场景。
在这种情况下，如果不做任何操作，那么新加入的服务器作为 Follower 服务器，其上的数据与 ZooKeeper 集群中其他服务器上的数据不一致。
当有新的查询会话请求发送到 ZooKeeper 集群进行处理，而恰巧该请求实际被分发给这台新加入的 Follower 机器进行处理，就会导致明明在集群中存在的数据，在这台服务器上却查询不到，导致数据查询不一致的情况。
因此，在当有新的 Follower 服务器加入 ZooKeeper 集群中的时候，该台服务器会在恢复模式下启动，并找到集群中的 Leader 节点服务器，并同该 Leader 服务器进行数据同步。
 
 

代码实现

LearnerHandler

public void run() {

  boolean needSnap = syncFollower(peerLastZxid, leader.zk.getZKDatabase(), leader);

  if(needSnap){

    leader.zk.getZKDatabase().serializeSnapshot(oa);

    oa.writeString("BenWasHere", "signature");

    bufferedOutput.flush();

  }

}

Leader 服务器选举的实现方法

服务启动时的 Leader 选举

发起投票

接收投票

统计投票

服务运行时的 Leader 选举

变更状态

发起投票

底层实现

public class QuorumCnxManager {

  final ConcurrentHashMap<Long, SendWorker> senderWorkerMap;

final ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>> queueSendMap;

final ConcurrentHashMap<Long, ByteBuffer> lastMessageSent;

}

class SendWorker extends ZooKeeperThread {

  Long sid;

  Socket sock;

  volatile boolean running = true;

  DataInputStream din;

  final SendWorker sw;

  public void run() {

  threadCnt.incrementAndGet();

  while (running && !shutdown && sock != null) {

    int length = din.readInt();

    if (length <= 0 || length > PACKETMAXSIZE) {

        throw new IOException(

                "Received packet with invalid packet: "

                        + length);

    }

    

    byte[] msgArray = new byte[length];

    din.readFully(msgArray, 0, length);

    ByteBuffer message = ByteBuffer.wrap(msgArray);

    addToRecvQueue(new Message(message.duplicate(), sid));

  }
  }
}

class SendWorker extends ZooKeeperThread {

  Long sid;

  Socket sock;

  RecvWorker recvWorker;

  volatile boolean running = true;

  DataOutputStream dout;

  public void run() {

    while (running && !shutdown && sock != null) {

    ByteBuffer b = null;

    try {

        ArrayBlockingQueue<ByteBuffer> bq = queueSendMap

                .get(sid);

        if (bq != null) {

            b = pollSendQueue(bq, 1000, TimeUnit.MILLISECONDS);

        } else {

            LOG.error("No queue of incoming messages for " +

                      "server " + sid);

            break;

        }

        if(b != null){

            lastMessageSent.put(sid, b);

            send(b);

        }

    } catch (InterruptedException e) {

        LOG.warn("Interrupted while waiting for message on queue",

                e);

    }

}

  }

}

public class FastLeaderElection implements Election {

  final static int maxNotificationInterval = 60000;

  final static int IGNOREVALUE = -1

  QuorumCnxManager manager;

}

synchronized public Vote getVote(){

    return new Vote(proposedLeader, proposedZxid, proposedEpoch);

 }

总结

Leader 与 Follower 的数据同步策略

同步方法

同步条件

同步过程

DIFF 同步

TRUNC+DIFF 同步

TRUNC 同步

SNAP 同步

同步后的处理

底层实现

public class Learner {}

protected void syncWithLeader(long newLeaderZxid) throws Exception{

 if (qp.getType() == Leader.DIFF) {

    snapshotNeeded = false;

  }else if (qp.getType() == Leader.TRUNC) {



  }

}

if (pif.hdr.getZxid() == qp.getZxid() && qp.getType() == Leader.COMMITANDACTIVATE) {

    QuorumVerifier qv = self.configFromString(new String(((SetDataTxn) pif.rec).getData()));

    boolean majorChange = self.processReconfig(qv, ByteBuffer.wrap(qp.getData()).getLong(),

            qp.getZxid(), true);

    if (majorChange) {

        throw new Exception("changes proposed in reconfig");

    }

}

if (!writeToTxnLog) {

    if (pif.hdr.getZxid() != qp.getZxid()) {

        LOG.warn("Committing " + qp.getZxid() + ", but next proposal is " + pif.hdr.getZxid());

    } else {

        zk.processTxn(pif.hdr, pif.rec);

        packetsNotCommitted.remove();

    }

} else {

    packetsCommitted.add(qp.getZxid());

集群中 Leader 的作用

事务性请求处理

Leader 事务处理分析

预处理阶段

事务处理阶段

事务执行阶段

响应客户端：

事务处理底层实现

protected void pRequest(Request request){

...

  switch (request.type) {

  case OpCode.setData:

    SetDataRequest setDataRequest = new SetDataRequest();                

    pRequest2Txn(request.type, zks.getNextZxid(), request, setDataRequest, true);

    break;

   }

}

case OpCode.setData:

    zks.sessionTracker.checkSession(request.sessionId, request.getOwner());

    SetDataRequest setDataRequest = (SetDataRequest)record;

    if(deserialize)

        ByteBufferInputStream.byteBuffer2Record(request.request, setDataRequest);

    path = setDataRequest.getPath();

    validatePath(path, request.sessionId);

    nodeRecord = getRecordForPath(path);

    checkACL(zks, request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, path, null);

    int newVersion = checkAndIncVersion(nodeRecord.stat.getVersion(), setDataRequest.getVersion(), path);

    request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));

    nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());

    nodeRecord.stat.setVersion(newVersion);

    addChangeRecord(nodeRecord);

集群中 Follow 的作用

非事务性请求处理过程

底层实现

protected void setupRequestProcessors() {

    RequestProcessor finalProcessor = new FinalRequestProcessor(this);

    commitProcessor = new CommitProcessor(finalProcessor,

            Long.toString(getServerId()), true, getZooKeeperServerListener());

    commitProcessor.start();

    firstProcessor = new FollowerRequestProcessor(this, commitProcessor);

    ((FollowerRequestProcessor) firstProcessor).start();

    syncProcessor = new SyncRequestProcessor(this,

            new SendAckRequestProcessor((Learner)getFollower()));

    syncProcessor.start();

选举过程

Leader 失效发现

Leader 重新选举

Follow 角色变更

集群同步数据

选举底层实现

switch (rstate) {

case 0:

    ackstate = QuorumPeer.ServerState.LOOKING;

    break;

case 1:

    ackstate = QuorumPeer.ServerState.FOLLOWING;

    break;

case 2:

    ackstate = QuorumPeer.ServerState.LEADING;

    break;

case 3:

    ackstate = QuorumPeer.ServerState.OBSERVING;

    break;

default:

    continue;

static public class ToSend {

    static enum mType {crequest, challenge, notification, ack}

    ToSend(mType type,

            long leader,

            long zxid,

            long electionEpoch,

            ServerState state,

            long sid,

            long peerEpoch,

            byte[] configData) {

        this.leader = leader;

        this.zxid = zxid;

        this.electionEpoch = electionEpoch;

        this.state = state;

        this.sid = sid;

        this.peerEpoch = peerEpoch;

        this.configData = configData;

    }

Observer 的作用

Observer 介绍

1. 随着集群规模的变大，集群处理写入的性能反而下降。

2. ZooKeeper 集群无法做到跨域部署

底层实现

peerType=observer

ObserverZooKeeperServer(FileTxnSnapLog logFactory, QuorumPeer self, ZKDatabase zkDb) throws IOException {

    super(logFactory, self.tickTime, self.minSessionTimeout, self.maxSessionTimeout, zkDb, self);

    LOG.info("syncEnabled =" + syncRequestProcessorEnabled);

public void commitRequest(Request request) {     

    if (syncRequestProcessorEnabled) {

        // Write to txnlog and take periodic snapshot

        syncProcessor.processRequest(request);

    }

    commitProcessor.commit(request);

INFORM 消息

Observer 处理链

public void run() {

        try {

            while (!finished) {

                Request request = queuedRequests.take();

               	...

                switch (request.type) {

                case OpCode.sync:

                    zks.pendingSyncs.add(request);

                    zks.getObserver().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.getObserver().request(request);

                    break;

               	...

            }

        } 

        ...

    }

}

public void processRequest(Request request) {

    if (stopped) {

        return;

    }

    if (LOG.isDebugEnabled()) {

        LOG.debug("Processing request:: " + request);

    }

    queuedRequests.add(request);

    wakeup();

}

分布式一致性算法

分布式事务

二阶段提交

底层实现

两阶段提交的问题

性能问题

单点问题

异常中断

三阶段提交

底层实现

ZAB 协议算法：崩溃恢复和消息广播

崩溃恢复

选票结构

• logicClock：用来记录服务器的投票轮次。logicClock 会从 1 开始计数，每当该台服务经过一轮投票后，logicClock 的数值就会加 1 。

• state：用来标记当前服务器的状态。在 ZooKeeper 集群中一台服务器具有 LOOKING、FOLLOWING、LEADERING、OBSERVING 这四种状态。

• self_id：用来表示当前服务器的 ID 信息，该字段在 ZooKeeper 集群中主要用来作为服务器的身份标识符。

• self_zxid： 当前服务器上所保存的数据的最大事务 ID ，从 0 开始计数。

• vote_id：投票要被推举的服务器的唯一 ID 。

• vote_zxid：被推举的服务器上所保存的数据的最大事务 ID ，从 0 开始计数。

选票筛选

消息广播

ZAB 与 Paxos 算法的联系与区别

Paxos 算法

底层实现

三种角色

• 提议者（Proposer）：提出提案（Proposal）。Proposal 信息包括提案编号（Proposal ID）和提议的值（Value）。

• 决策者（Acceptor）：参与决策，回应 Proposers 的提案。收到 Proposal 后可以接受提案，若 Proposal 获得超过半数 Acceptors 的许可，则称该 Proposal 被批准。

• 决策学习者：不参与决策，从 Proposers/Acceptors 学习最新达成一致的提案（Value）。

事务处理过程

• 提案准备阶段：该阶段是整个 Paxos 算法的最初阶段，所有接收到的来自客户端的事务性会话在执行之前，整个集群中的 Proposer 角色服务器或者节点，需要将会话发送给 Acceptor 决策者服务器。在 Acceptor 服务器接收到该条询问信息后，需要返回 Promise ，承诺可以执行操作信息给 Proposer 角色服务器。

• 事务处理阶段：在经过提案准备阶段，确认该条事务性的会话操作可以在集群中正常执行后，Proposer 提案服务器会再次向 Acceptor 决策者服务器发送 propose 提交请求。Acceptor 决策者服务器在接收到该 propose 请求后，在本地执行该条事务性的会话操作。

• 数据同步阶段：在完成了事务处理阶段的操作后，整个集群中对该条事务性会话的数据变更已经在 Acceptor 决策者服务器上执行完成，当整个集群中有超过半数的 Acceptor 决策者服务器都成功执行后，Paxos 算法将针对本次执行结果形成一个决议，并发送给 Learner 服务器。当 Learner 服务器接收到该条决议信息后，会同步 Acceptor 决策者服务器上的数据信息，最终完成该条事务性会话在整个集群中的处理。

Paxos PK ZAB

ZooKeeper 中二阶段提交算法的实现分析

提交请求

执行请求

底层实现

public class SendAckRequestProcessor implements RequestProcessor, Flushable { 

  public void processRequest(Request si) { 

    if(si.type != OpCode.sync){ 

        QuorumPacket qp = new QuorumPacket(Leader.ACK, si.getHdr().getZxid(), null, 

            null); 

        try { 

            learner.writePacket(qp, false); 

        } catch (IOException e) { 

            LOG.warn("Closing connection to leader, exception during packet send", e); 

            try { 

                if (!learner.sock.isClosed()) { 

                    learner.sock.close(); 

                } 

            } catch (IOException e1) { 

                // Nothing to do, we are shutting things down, so an exception here is irrelevant 

                LOG.debug("Ignoring error closing the connection", e1); 

            } 

        } 

    } 

} 

}

public ProposalRequestProcessor(LeaderZooKeeperServer zks, 

        RequestProcessor nextProcessor) { 

    this.zks = zks; 

    this.nextProcessor = nextProcessor; 

    AckRequestProcessor ackProcessor = new AckRequestProcessor(zks.getLeader()); 

    syncProcessor = new SyncRequestProcessor(zks, ackProcessor); 

}

class AckRequestProcessor implements RequestProcessor { 

public void processRequest(Request request) { 

    QuorumPeer self = leader.self; 

    if(self != null) 

        leader.processAck(self.getId(), request.zxid, null); 

    else 

        LOG.error("Null QuorumPeer"); 

}}

Raft 算法

实现过程

强领导者

领导选举

成员关系