[[391275]]

Zookeeper 是通過(guò) ZAB 一致性協(xié)議來(lái)實(shí)現(xiàn)分布式事務(wù)的最終一致性。

ZAB 協(xié)議介紹

ZAB 全稱為 Zookeeper Atomic Broadcast(Zookeeper 原子廣播協(xié)議)

ZAB 協(xié)議是為分布式協(xié)調(diào)服務(wù)ZooKeeper專門(mén)設(shè)計(jì)的一種支持崩潰恢復(fù)的一致性協(xié)議。基于該協(xié)議，ZooKeeper 實(shí)現(xiàn)了一種主從模式的系統(tǒng)架構(gòu)來(lái)保持集群中各個(gè)副本之間的數(shù)據(jù)一致性。

ZAB的消息廣播過(guò)程使用的是原子廣播協(xié)議，類(lèi)似于二階段提交。針對(duì)客戶端的請(qǐng)求，Leader服務(wù)器生成對(duì)應(yīng)的事務(wù)提議，并將其發(fā)送給集群中所有的 Follower 服務(wù)器。然后收集各自的選票，最后進(jìn)行事務(wù)提交。如圖：

在 ZAB 協(xié)議中二階段提交，移除了中斷邏輯。所有的 Follower 服務(wù)器要么正常反饋 Leader 提出的事務(wù)提議，要么就拋棄 Leader 服務(wù)器。同時(shí)，我們可以在過(guò)半的 Follower 服務(wù)器已經(jīng)反饋 ACK 后，就開(kāi)始提交事務(wù)提議了。

Leader 服務(wù)器會(huì)為事務(wù)提議分配一個(gè)全局單調(diào)遞增的 ID，稱為事務(wù) ID(ZXID)。由于 ZAB 協(xié)議需要保證每一個(gè)消息嚴(yán)格的因果關(guān)系，因此需要將每一個(gè)事務(wù)提議按照其 ZXID 的先后順序進(jìn)行處理。

在消息廣播過(guò)程中，Leader 服務(wù)器會(huì)為每一個(gè) Follower 服務(wù)器分配一個(gè)隊(duì)列，然后將事務(wù)提議依次放入到這些隊(duì)列中去，并且根據(jù) FIFO 的策略進(jìn)行消息發(fā)送。

每一個(gè) Follower 服務(wù)器接收到這個(gè)事務(wù)提議后，會(huì)把該事務(wù)提議以事務(wù)日志的形式寫(xiě)入到本地磁盤(pán)中，并且寫(xiě)入成功后，反饋給 Leader 服務(wù)器 ACK。

當(dāng) Leader 服務(wù)器收到過(guò)半 Follower 服務(wù)器的 ACK，就發(fā)送一個(gè) COMMIT 消息，同時(shí) Leader 自身完成事務(wù)提交，F(xiàn)ollower 服務(wù)器接收到 COMMIT 消息后，也進(jìn)行事務(wù)提交。

之所以采用原子廣播協(xié)議協(xié)議，是為了保證分布式數(shù)據(jù)一致性。過(guò)半的節(jié)點(diǎn)數(shù)據(jù)保存一致性。

消息廣播

你可以認(rèn)為消息廣播機(jī)制是簡(jiǎn)化版的 2PC協(xié)議，就是通過(guò)如下的機(jī)制保證事務(wù)的順序一致性的。

客戶端提交事務(wù)請(qǐng)求時(shí) Leader 節(jié)點(diǎn)為每一個(gè)請(qǐng)求生成一個(gè)事務(wù) Proposal，將其發(fā)送給集群中所有的 Follower 節(jié)點(diǎn)，收到過(guò)半 Follower的反饋后開(kāi)始對(duì)事務(wù)進(jìn)行提交，ZAB 協(xié)議使用了原子廣播協(xié)議;在 ZAB 協(xié)議中只需要得到過(guò)半的 Follower 節(jié)點(diǎn)反饋 Ack 就可以對(duì)事務(wù)進(jìn)行提交，這也導(dǎo)致了 Leader 節(jié)點(diǎn)崩潰后可能會(huì)出現(xiàn)數(shù)據(jù)不一致的情況，ZAB 使用了崩潰恢復(fù)來(lái)處理數(shù)字不一致問(wèn)題;消息廣播使用了TCP 協(xié)議進(jìn)行通訊所有保證了接受和發(fā)送事務(wù)的順序性。廣播消息時(shí) Leader 節(jié)點(diǎn)為每個(gè)事務(wù) Proposal分配一個(gè)全局遞增的 ZXID(事務(wù)ID)，每個(gè)事務(wù) Proposal 都按照 ZXID 順序來(lái)處理;

Leader 節(jié)點(diǎn)為每一個(gè) Follower 節(jié)點(diǎn)分配一個(gè)隊(duì)列按事務(wù) ZXID 順序放入到隊(duì)列中，且根據(jù)隊(duì)列的規(guī)則 FIFO 來(lái)進(jìn)行事務(wù)的發(fā)送。Follower節(jié)點(diǎn)收到事務(wù) Proposal 后會(huì)將該事務(wù)以事務(wù)日志方式寫(xiě)入到本地磁盤(pán)中，成功后反饋 Ack 消息給 Leader 節(jié)點(diǎn)，Leader 在接收到過(guò)半Follower 節(jié)點(diǎn)的 Ack 反饋后就會(huì)進(jìn)行事務(wù)的提交，以此同時(shí)向所有的 Follower 節(jié)點(diǎn)廣播 Commit 消息，F(xiàn)ollower 節(jié)點(diǎn)收到 Commit 后開(kāi)始對(duì)事務(wù)進(jìn)行提交;

崩潰恢復(fù)

消息廣播過(guò)程中，Leader 崩潰了還能保證數(shù)據(jù)一致嗎?當(dāng) Leader 崩潰會(huì)進(jìn)入崩潰恢復(fù)模式。其實(shí)主要是對(duì)如下兩種情況的處理。

1. Leader 在復(fù)制數(shù)據(jù)給所有 Follwer 之后崩潰，怎么處理?
2. Leader 在收到 Ack 并提交了自己，同時(shí)發(fā)送了部分 commit 出去之后崩潰，怎么處理?

針對(duì)此問(wèn)題，ZAB 定義了 2 個(gè)原則：

1. ZAB 協(xié)議確保執(zhí)行那些已經(jīng)在 Leader 提交的事務(wù)最終會(huì)被所有服務(wù)器提交。
2. ZAB 協(xié)議確保丟棄那些只在 Leader 提出/復(fù)制，但沒(méi)有提交的事務(wù)。

至于如何實(shí)現(xiàn)確保提交已經(jīng)被 Leader 提交的事務(wù)，同時(shí)丟棄已經(jīng)被跳過(guò)的事務(wù)呢?核心是通過(guò) ZXID 來(lái)進(jìn)行處理。在崩潰過(guò)后進(jìn)行恢復(fù)的時(shí)候會(huì)選擇最大的 zxid 作為恢復(fù)的快照。這樣的好處是: 可以省略事務(wù)提交的檢查和事務(wù)的丟棄工作以提升效率

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

完成Leader選舉之后，在正式開(kāi)始工作之前，Leader服務(wù)器會(huì)去確認(rèn)事務(wù)日志中所有事務(wù)提議(指已經(jīng)提交的事務(wù)提議)是否都已經(jīng)被過(guò)半的機(jī)器提交了，即是否完成數(shù)據(jù)同步。下面是ZAB協(xié)議的 數(shù)據(jù)同步過(guò)程。

Leader服務(wù)器為每一個(gè)Follower服務(wù)器準(zhǔn)備一個(gè)隊(duì)列，將那些沒(méi)有被Follower服務(wù)器同步的事務(wù)以事務(wù)提議的形式逐個(gè)發(fā)送給Follower服務(wù)器，并在每一個(gè)事務(wù)提議消息后面發(fā)送一個(gè)commit消息，表示該事務(wù)已被提交。

等到Follower服務(wù)器將所有其未同步的事務(wù)提議都從Leader服務(wù)器上面同步過(guò)來(lái)，并且應(yīng)用到本地?cái)?shù)據(jù)庫(kù)后，Leader服務(wù)器就會(huì)將該Follower服務(wù)器加入到真正可用的Follower列表中。

ZXID 的設(shè)計(jì)

ZXID 是一個(gè)64位的數(shù)字, 如下圖所示。

其中低 32 位是一個(gè)簡(jiǎn)單的單調(diào)遞增的計(jì)數(shù)器，Leader 服務(wù)器產(chǎn)生一個(gè)新的事務(wù)提議的時(shí)候，都會(huì)對(duì)該計(jì)數(shù)器 +1。

高 32 位，用來(lái)區(qū)分不同的 Leader 服務(wù)器。具體做法是，每選舉產(chǎn)生一個(gè)新的 Leader 服務(wù)器，就會(huì)從 Leader 服務(wù)器的本地日志中取出一個(gè)最大的 ZXID，生成對(duì)應(yīng)的 epoch 值，然后再進(jìn)行加1操作，之后就會(huì)以該值作為新的 epoch。并將低 32 位從 0 開(kāi)始生成 ZXID。(我理解這里的 epoch 代表的就是一個(gè) Leader 服務(wù)器的標(biāo)志，每次選舉 Leader 服務(wù)器，那么 epoch 值就會(huì)更新，代表是這段時(shí)期由這個(gè)新的 Leader 服務(wù)器進(jìn)行事務(wù)請(qǐng)求的處理)。

ZAB 協(xié)議中通過(guò) epoch 編號(hào)來(lái)區(qū)分 Leader 周期變化，能夠有效避免不同 Leader 服務(wù)器使用相同的 ZXID。

下面是我 Leader 節(jié)點(diǎn)的 zxid 生成核心代碼大家可以看一下。

// Leader.java 
void lead() throws IOException, InterruptedException { 
    // .... 
  long epoch = getEpochToPropose(self.getId(), self.getAcceptedEpoch()); 
  zk.setZxid(ZxidUtils.makeZxid(epoch, 0)); 
  // .... 
} 
// 
public long getEpochToPropose(long sid, long lastAcceptedEpoch) throws InterruptedException, IOException { 
  synchronized (connectingFollowers) { 
    // .... 
    if (isParticipant(sid)) { 
      // 將自己加入連接隊(duì)伍中，方便后面判斷 lead 是否有效 
      connectingFollowers.add(sid); 
    } 
    QuorumVerifier verifier = self.getQuorumVerifier(); 
    // 如果有足夠多的 follower 進(jìn)入， 選舉有效，則無(wú)需等待，并通過(guò)其他等待的線程，類(lèi)似 Barrier 
    if (connectingFollowers.contains(self.getId()) && verifier.containsQuorum(connectingFollowers)) { 
      waitingForNewEpoch = false; 
      self.setAcceptedEpoch(epoch); 
      connectingFollowers.notifyAll(); 
    } else { 
      // .... 
      // followers 不夠就進(jìn)入等待， 超時(shí)時(shí)間為 initLimit 
      while (waitingForNewEpoch && cur < end && !quitWaitForEpoch) { 
        connectingFollowers.wait(end - cur); 
        cur = Time.currentElapsedTime(); 
      } 
      // 超時(shí)退出，重新選舉 
      if (waitingForNewEpoch) { 
        throw new InterruptedException("Timeout while waiting for epoch from quorum"); 
      } 
    } 
    return epoch; 
  } 
} 
// ZxidUtils 
public static long makeZxid(long epoch, long counter) { 
  return (epoch << 32L) | (counter & 0xffffffffL); 
} 

ZAB 協(xié)議實(shí)現(xiàn)

寫(xiě)數(shù)據(jù)的過(guò)程

下面我梳理了 zookeeper 源碼中寫(xiě)數(shù)據(jù)的過(guò)程，如下圖所示：

參考資料

https://www.cnblogs.com/veblen/p/10985676.html

https://zookeeper.apache.org