面試官上來就問 ZAB 協議，瑟瑟發抖…
Zookeeper 是通過 ZAB 一致性協議來實現分布式事務的最終一致性。
ZAB 協議介紹
ZAB 全稱為 Zookeeper Atomic Broadcast(Zookeeper 原子廣播協議)
ZAB 協議是為分布式協調服務ZooKeeper專門設計的一種支持崩潰恢復的一致性協議。基于該協議，ZooKeeper 實現了一種主從模式的系統架構來保持集群中各個副本之間的數據一致性。
ZAB的消息廣播過程使用的是原子廣播協議，類似于二階段提交。針對客戶端的請求，Leader服務器生成對應的事務提議，并將其發送給集群中所有的 Follower 服務器。然后收集各自的選票，最后進行事務提交。如圖：
在 ZAB 協議中二階段提交，移除了中斷邏輯。所有的 Follower 服務器要么正常反饋 Leader 提出的事務提議，要么就拋棄 Leader 服務器。同時，我們可以在過半的 Follower 服務器已經反饋 ACK 后，就開始提交事務提議了。
Leader 服務器會為事務提議分配一個全局單調遞增的 ID，稱為事務 ID(ZXID)。由于 ZAB 協議需要保證每一個消息嚴格的因果關系，因此需要將每一個事務提議按照其 ZXID 的先后順序進行處理。
在消息廣播過程中，Leader 服務器會為每一個 Follower 服務器分配一個隊列，然后將事務提議依次放入到這些隊列中去，并且根據 FIFO 的策略進行消息發送。
每一個 Follower 服務器接收到這個事務提議后，會把該事務提議以事務日志的形式寫入到本地磁盤中，并且寫入成功后，反饋給 Leader 服務器 ACK。
當 Leader 服務器收到過半 Follower 服務器的 ACK，就發送一個 COMMIT 消息，同時 Leader 自身完成事務提交，Follower 服務器接收到 COMMIT 消息后，也進行事務提交。
之所以采用原子廣播協議協議，是為了保證分布式數據一致性。過半的節點數據保存一致性。
消息廣播
你可以認為消息廣播機制是簡化版的 2PC協議，就是通過如下的機制保證事務的順序一致性的。
客戶端提交事務請求時 Leader 節點為每一個請求生成一個事務 Proposal，將其發送給集群中所有的 Follower 節點，收到過半 Follower的反饋后開始對事務進行提交，ZAB 協議使用了原子廣播協議;在 ZAB 協議中只需要得到過半的 Follower 節點反饋 Ack 就可以對事務進行提交，這也導致了 Leader 節點崩潰后可能會出現數據不一致的情況，ZAB 使用了崩潰恢復來處理數字不一致問題;消息廣播使用了TCP 協議進行通訊所有保證了接受和發送事務的順序性。廣播消息時 Leader 節點為每個事務 Proposal分配一個全局遞增的 ZXID(事務ID)，每個事務 Proposal 都按照 ZXID 順序來處理;
Leader 節點為每一個 Follower 節點分配一個隊列按事務 ZXID 順序放入到隊列中，且根據隊列的規則 FIFO 來進行事務的發送。Follower節點收到事務 Proposal 后會將該事務以事務日志方式寫入到本地磁盤中，成功后反饋 Ack 消息給 Leader 節點，Leader 在接收到過半Follower 節點的 Ack 反饋后就會進行事務的提交，以此同時向所有的 Follower 節點廣播 Commit 消息，Follower 節點收到 Commit 后開始對事務進行提交;
崩潰恢復
消息廣播過程中，Leader 崩潰了還能保證數據一致嗎?當 Leader 崩潰會進入崩潰恢復模式。其實主要是對如下兩種情況的處理。
- Leader 在復制數據給所有 Follwer 之后崩潰，怎么處理?
- Leader 在收到 Ack 并提交了自己，同時發送了部分 commit 出去之后崩潰，怎么處理?
針對此問題，ZAB 定義了 2 個原則：
- ZAB 協議確保執行那些已經在 Leader 提交的事務最終會被所有服務器提交。
- ZAB 協議確保丟棄那些只在 Leader 提出/復制，但沒有提交的事務。
至于如何實現確保提交已經被 Leader 提交的事務，同時丟棄已經被跳過的事務呢?核心是通過 ZXID 來進行處理。在崩潰過后進行恢復的時候會選擇最大的 zxid 作為恢復的快照。這樣的好處是: 可以省略事務提交的檢查和事務的丟棄工作以提升效率
數據同步
完成Leader選舉之后，在正式開始工作之前，Leader服務器會去確認事務日志中所有事務提議(指已經提交的事務提議)是否都已經被過半的機器提交了，即是否完成數據同步。下面是ZAB協議的 數據同步過程。
Leader服務器為每一個Follower服務器準備一個隊列，將那些沒有被Follower服務器同步的事務以事務提議的形式逐個發送給Follower服務器，并在每一個事務提議消息后面發送一個commit消息，表示該事務已被提交。
等到Follower服務器將所有其未同步的事務提議都從Leader服務器上面同步過來，并且應用到本地數據庫后，Leader服務器就會將該Follower服務器加入到真正可用的Follower列表中。
ZXID 的設計
ZXID 是一個64位的數字, 如下圖所示。
其中低 32 位是一個簡單的單調遞增的計數器，Leader 服務器產生一個新的事務提議的時候，都會對該計數器 +1。
高 32 位，用來區分不同的 Leader 服務器。具體做法是，每選舉產生一個新的 Leader 服務器，就會從 Leader 服務器的本地日志中取出一個最大的 ZXID，生成對應的 epoch 值，然后再進行加1操作，之后就會以該值作為新的 epoch。并將低 32 位從 0 開始生成 ZXID。(我理解這里的 epoch 代表的就是一個 Leader 服務器的標志，每次選舉 Leader 服務器，那么 epoch 值就會更新，代表是這段時期由這個新的 Leader 服務器進行事務請求的處理)。
ZAB 協議中通過 epoch 編號來區分 Leader 周期變化，能夠有效避免不同 Leader 服務器使用相同的 ZXID。
下面是我 Leader 節點的 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 協議實現
寫數據的過程
下面我梳理了 zookeeper 源碼中寫數據的過程，如下圖所示：
參考資料
https://www.cnblogs.com/veblen/p/10985676.html
https://zookeeper.apache.org
