筆者關(guān)注ZooKeeper有一段時(shí)間，從ZooKeeper提供的API中，發(fā)現(xiàn)有一個(gè)比較有意思的API，叫sync。但一直不太明白sync API的含義。

從官方的文檔里，找到了這段話：

Sometimes developers mistakenly assume one other guarantee that ZooKeeper does not in fact make. This is:

Simultaneously Consistent Cross-Client Views
ZooKeeper does not guarantee that at every instance in time, two different clients will have identical views of ZooKeeper data. Due to factors like network delays, one client may perform an update before another client gets notified of the change. Consider the scenario of two clients, A and B. If client A sets the value of a znode /a from 0 to 1, then tells client B to read /a, client B may read the old value of 0, depending on which server it is connected to. If it is important that Client A and Client B read the same value, Client B should should call the sync() method from the ZooKeeper API method before it performs its read.

So, ZooKeeper by itself doesn't guarantee that changes occur synchronously across all servers, but ZooKeeper primitives can be used to construct higher level functions that provide useful client synchronization.

于是我們知道，sync是使得client當(dāng)前連接著的ZooKeeper服務(wù)器，和ZooKeeper的Leader節(jié)點(diǎn)同步（sync）一下數(shù)據(jù)。

另外我也同時(shí)看了一下源代碼，當(dāng)follower收到到sync請(qǐng)求時(shí)，會(huì)將這個(gè)請(qǐng)求添加到一個(gè)pendingSyncs隊(duì)列里，然后將這個(gè)請(qǐng)求發(fā)送給leader，直到收到leader的Leader.SYNC消息時(shí)，才將這個(gè)請(qǐng)求從pendingSyncs隊(duì)列里移除，并commit這個(gè)請(qǐng)求。

FollowerRequestProcessor.java:

@Override
    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:
                    // 可以看到sync請(qǐng)求和其它事務(wù)型請(qǐng)求的的區(qū)別在于，除了發(fā)送給leader之外，還要記錄到pendingSyncs里
                    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) {
            handleException(this.getName(), e);
        }
        LOG.info("FollowerRequestProcessor exited loop!");
    }

FollowerZooKeeperServer.java:

    // 接收到leader發(fā)送的Leader.SYNC消息后，才真正commit這個(gè)請(qǐng)求
    synchronized public void sync(){
        if(pendingSyncs.size() ==0){
             LOG.warn("Not expecting a sync.");
            return;
        }
                
        Request r = pendingSyncs.remove();
        commitProcessor.commit(r);
    }

當(dāng)Leader收到一個(gè)sync請(qǐng)求時(shí)，如果leader當(dāng)前沒有待commit的決議，那么leader會(huì)立即發(fā)送一個(gè)Leader.SYNC消息給follower。否則，leader會(huì)等到當(dāng)前最后一個(gè)待commit的決議完成后，再發(fā)送Leader.SYNC消息給Follower。

Leader.java:

    synchronized public void processSync(LearnerSyncRequest r){
        if(outstandingProposals.isEmpty()){
            sendSync(r);
        } else {
            List<LearnerSyncRequest> l = pendingSyncs.get(lastProposed);
            if (l == null) {
                l = new ArrayList<LearnerSyncRequest>();
            }
            l.add(r);
            pendingSyncs.put(lastProposed, l);
        }
    }

其實(shí)這里面有一個(gè)隱含的邏輯，就是如果leader和follower之間的消息通信，是嚴(yán)格按順序來發(fā)送的（TCP保證），因此，當(dāng)follower接收到Leader.SYNC消息時(shí)，說明follower也一定接收到了leader之前（在leader接收到sync請(qǐng)求之前）發(fā)送的所有提案或者commit消息。這樣，就可以確保follower和leader是同步的了。