[[391275]]

Zookeeper 是通過 ZAB 一致性協(xié)議來實現(xiàn)分布式事務的最終一致性。

ZAB 協(xié)議介紹

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

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

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

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

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

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

每一個 Follower 服務器接收到這個事務提議后，會把該事務提議以事務日志的形式寫入到本地磁盤中，并且寫入成功后，反饋給 Leader 服務器 ACK。

當 Leader 服務器收到過半 Follower 服務器的 ACK，就發(fā)送一個 COMMIT 消息，同時 Leader 自身完成事務提交，F(xiàn)ollower 服務器接收到 COMMIT 消息后，也進行事務提交。

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

消息廣播

你可以認為消息廣播機制是簡化版的 2PC協(xié)議，就是通過如下的機制保證事務的順序一致性的。

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

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

崩潰恢復

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

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

針對此問題，ZAB 定義了 2 個原則：

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

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

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

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

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

等到Follower服務器將所有其未同步的事務提議都從Leader服務器上面同步過來，并且應用到本地數(shù)據(jù)庫后，Leader服務器就會將該Follower服務器加入到真正可用的Follower列表中。

ZXID 的設計

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

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

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

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

下面是我 Leader 節(jié)點的 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)) { 
      // 將自己加入連接隊伍中，方便后面判斷 lead 是否有效 
      connectingFollowers.add(sid); 
    } 
    QuorumVerifier verifier = self.getQuorumVerifier(); 
    // 如果有足夠多的 follower 進入， 選舉有效，則無需等待，并通過其他等待的線程，類似 Barrier 
    if (connectingFollowers.contains(self.getId()) && verifier.containsQuorum(connectingFollowers)) { 
      waitingForNewEpoch = false; 
      self.setAcceptedEpoch(epoch); 
      connectingFollowers.notifyAll(); 
    } else { 
      // .... 
      // followers 不夠就進入等待， 超時時間為 initLimit 
      while (waitingForNewEpoch && cur < end && !quitWaitForEpoch) { 
        connectingFollowers.wait(end - cur); 
        cur = Time.currentElapsedTime(); 
      } 
      // 超時退出，重新選舉 
      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é)議實現(xiàn)

寫數(shù)據(jù)的過程

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

參考資料

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

https://zookeeper.apache.org