分布式協調框架Zookeeper核心設計理解與實戰

作者： wangkai 2021-07-29 07:48:36

想起很久以前在某個客戶現場，微服務 B 突然無法調用到微服務 A，為了使服務盡快正常恢復，重啟了微服務 B 。雖然故障原因找到了，但對于 Zookeeper 的理解還是不夠深刻，于是重新學習了 Zookeeper 的核心設計，并記錄于此文共勉。

[[413943]]

本文轉載自微信公眾號「KK架構」，作者wangkai 。轉載本文請聯系KK架構公眾號。

一、前言

想起很久以前在某個客戶現場，微服務 B 突然無法調用到微服務 A，為了使服務盡快正常恢復，重啟了微服務 B 。

但客戶不依不饒詢問這個問題出現的原因，于是我還大老遠從杭州飛到深圳，現場排查問題。

最后的結論是，zk 在某時刻出現主備切換，此時微服務 A(基于 dubbo)需要重新往 zk上注冊，但是端口號變了。

但是微服務 B 本地有微服務 A rpc 接口的緩存，緩存里面還是舊的端口，所以調用不到。

解決方法就是，把微服務的 rpc 端口號改成固定的。

雖說原因找到了，但對于 Zookeeper 的理解還是不夠深刻，于是重新學習了 Zookeeper 的核心設計，并記錄于此文共勉。

二、Zookeeper 核心架構設計

1、Zookeeper 特點

(1)Zookeeper 是一個分布式協調服務，是為了解決多個節點狀態不一致的問題，充當中間機構來調停。如果出現了不一致，則把這個不一致的情況寫入到 Zookeeper 中，Zookeeper 會返回響應，響應成功，則表示幫你達成了一致。

比如，A、B、C 節點在集群啟動時，需要推舉出一個主節點，這個時候，A、B、C 只要同時往 Zookeeper 上注冊臨時節點，誰先注冊成功，誰就是主節點。

(2)Zookeeper 雖然是一個集群，但是數據并不是分散存儲在各個節點上的，而是每個節點都保存了集群所有的數據。

其中一個節點作為主節點，提供分布式事務的寫服務，其他節點和這個節點同步數據，保持和主節點狀態一致。

(3)Zookeeper 所有節點的數據狀態通過 Zab 協議保持一致。當集群中沒有 Leader 節點時，內部會執行選舉，選舉結束，Follower 和 Leader 執行狀態同步;當有 Leader 節點時，Leader 通過 ZAB 協議主導分布式事務的執行，并且所有的事務都是串行執行的。

(4)Zookeeper 的節點個數是不能線性擴展的，節點越多，同步數據的壓力越大，執行分布式事務性能越差。推薦3、5、7 這樣的數目。

2、Zookeeper 角色的理解

Zookeeper 并沒有沿用 Master/Slave 概念，而是引入了 Leader，Follower，Observer 三種角色。

通過 Leader 選舉算法來選定一臺服務器充當 Leader 節點，Leader 服務器為客戶端提供讀、寫服務。

Follower 節點可以參加選舉，也可以接受客戶端的讀請求，但是接受到客戶端的寫請求時，會轉發到 Leader 服務器去處理。

Observer 角色只能提供讀服務，不能選舉和被選舉，所以它存在的意義是在不影響寫性能的前提下，提升集群的讀性能。

3、Zookeeper 同時滿足了 CAP 嗎?

答案是否，CAP 只能同時滿足其二。

Zookeeper 是有取舍的，它實現了 A 可用性、P 分區容錯性、C 的寫入一致性，犧牲的是 C的讀一致性。

也就是說，Zookeeper 并不保證讀取的一定是最新的數據。如果一定要最新，需要使用 sync 回調處理。

三、核心機制一：ZNode 數據模型

Zookeeper 的 ZNode 模型其實可以理解為類文件系統，如下圖：

1、ZNode 并不適合存儲太大的數據

為什么是類文件系統呢?因為 ZNode 模型沒有文件和文件夾的概念，每個節點既可以有子節點，也可以存儲數據。

那么既然每個節點可以存儲數據，是不是可以任意存儲無限制的數據呢?答案是否定的。在 Zookeeper 中，限制了每個節點只能存儲小于 1 M 的數據，實際應用中，最好不要超過 1kb。

原因有以下四點：

同步壓力：Zookeeper 的每個節點都存儲了 Zookeeper 的所有數據，每個節點的狀態都要保持和 Leader 一致，同步過程至少要保證半數以上的節點同步成功，才算最終成功。如果數據越大，則寫入的難度也越大。
請求阻塞：Zookeeper 為了保證寫入的強一致性，會嚴格按照寫入的順序串行執行，某個時刻只能執行一個事務。如果上一個事務執行耗時比較長，會阻塞后面的請求;
存儲壓力：正是因為每個 Zookeeper 的節點都存儲了完整的數據，每個 ZNode 存儲的數據越大，則消耗的物理內存也越大;
設計初衷：Zookeeper 的設計初衷，不是為了提供大規模的存儲服務，而是提供了這樣的數據模型解決一些分布式問題。

2、ZNode 的分類

(1)按生命周期分類

按照聲明周期，ZNode 可分為永久節點和臨時節點。

很好理解，永久節點就是要顯示的刪除，否則會一直存在;臨時節點，是和會話綁定的，會話創建的所有節點，會在會話斷開連接時，全部被 Zookeeper 系統刪除。

(2)按照是否帶序列號分類

帶序列號的話，比如在代碼中創建 /a 節點，創建之后其實是 /a000000000000001，再創建的話，就是 /a000000000000002，依次遞增

不帶序號，就是創建什么就是什么

(3)所以，一共有四種 ZNode：

永久的不帶序號的
永久的帶序號的
臨時的不帶序號的
臨時的帶序號的

(4)注意的點

臨時節點下面不能掛載子節點，只能作為其他節點的葉子節點。

3. 代碼實戰

ZNode 的數據模型其實很簡單，只有這么多知識。下面用代碼來鞏固一下。

這里我們使用 curator 框架來做 demo。(當然，你可以選擇使用 Zookeeper 官方自帶的 Api)

引入 pom 坐標：

<!-- curator-framework --> 
<dependency> 
    <groupId>org.apache.curator</groupId> 
    <artifactId>curator-framework</artifactId> 
    <version>4.2.0</version> 
</dependency> 
<!-- curator-recipes --> 
<dependency> 
    <groupId>org.apache.curator</groupId> 
    <artifactId>curator-recipes</artifactId> 
    <version>4.2.0</version> 
</dependency>

代碼：

public class ZkTest { 
 
    // 會話超時 
    private final int SESSION_TIMEOUT = 30 * 1000; 
 
    // 連接超時 、 有啥區別 
    private static final int CONNECTION_TIMEOUT = 3 * 1000; 
 
    private static final String CONNECT_ADDR = "localhost:2181"; 
 
    private CuratorFramework client = null; 
 
    public static void main(String[] args) throws Exception { 
        // 創建客戶端 
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10); 
        CuratorFramework client = CuratorFrameworkFactory.builder() 
                .connectString(CONNECT_ADDR) 
                .connectionTimeoutMs(CONNECTION_TIMEOUT) 
                .retryPolicy(retryPolicy) 
                .build(); 
        client.start(); 
        System.out.println(ZooKeeper.States.CONNECTED); 
        System.out.println(client.getState()); 
 
        // 創建節點 /test1 
        client.create() 
                .forPath("/test1", "curator data".getBytes(StandardCharsets.UTF_8)); 
 
        System.out.println(client.getChildren().forPath("/")); 
 
        // 臨時節點 
        client.create().withMode(CreateMode.EPHEMERAL) 
                .forPath("/secondPath", "hello world".getBytes(StandardCharsets.UTF_8)); 
        System.out.println(new String(client.getData().forPath("/secondPath"))); 
 
        client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL) 
                .forPath("/abc", "hello".getBytes(StandardCharsets.UTF_8)); 
        // 遞歸創建 
        client.create() 
                .creatingParentContainersIfNeeded() 
                .forPath("/secondPath1/sencond2/sencond3"); 
 
 
        Thread.sleep(10000); 
    }

四、核心機制二：Watcher 監聽機制

Watcher 監聽機制是 Zookeeper 解決各種分布式不一致疑難雜癥的獨家法門，也是學習 Zookeeper 必學的知識點。

1. 對于 Watcher 機制的理解

Zookeeper 提供了數據的發布與訂閱的功能，多個訂閱者可以同時監聽某一個對象，當這個對象自身狀態發生變化時(例如節點數據或者節點的子節點個數變化)，Zookeeper 系統會通知這些訂閱者。

對于發布和訂閱這個概念的理解，我們可以用這個場景來理解：

比如前兩天的臺風，老板想發一個通知給員工：明天在家辦公。

于是老板會在釘釘群上 Ding 一個消息，員工自己打開釘釘查看。

在這個場景中，老板是發布者，員工是訂閱者，釘釘群就是 Zookeeper 系統。

老板并不一一給員工發消息，而是把消息發到群里，員工就可以感知到消息的變化。

訂閱者	員工	客戶端1
系統	釘釘群	Zookeeper系統
發布者	老板	客戶端2

2、 Watcher 機制的流程

客戶端首先將 Watcher 注冊到服務器上，同時將 Watcher 對象保存在客戶端的 Watcher 管理器中。當 Zookeeper 服務端監聽到數據狀態發生變化時，服務端會首先主動通知客戶端，接著客戶端的 Watcher 管理器會觸發相關的 Watcher 來回調響應的邏輯，從而完成整體的發布/訂閱流程。

監聽器 Watcher 的定義：

public interface Watcher { 
//   WatchedEvent 對象中有下面三個屬性，Zookeeper狀態，事件類型，路徑 
//    final private KeeperState keeperState; 
//    final private EventType eventType; 
//    private String path; 
    abstract public void process(WatchedEvent event); 
}

下面是監聽的大致流程圖：

稍稍解釋一下：

1、Client1 和 Client2 都關心 /app2 節點的數據狀態變化，于是注冊一個對于 /app2 的監聽器到 Zookeeper 上;

2、當 Client3 修改 /app2 的值后，Zookeeper 會主動通知 Client1 和 Client2 ，并且回調監聽器的方法。

當然這里的數據狀態變化有下面這些類型：

節點被創建;
節點被刪除;
節點數據發生改變;
節點的子節點個數發生改變。

3. 通過代碼來初步理解

我們還是用 Curator 框架來驗證一下這個監聽器。

代碼很簡單，這里我們使用 TreeCache 表示對于 /app2 的監聽，并且注冊了監聽的方法。

public class CuratorWatcher { 
 
    public static void main(String[] args) throws Exception { 
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181") 
                .connectionTimeoutMs(10000) 
                .retryPolicy(new ExponentialBackoffRetry(1000, 10)) 
                .build(); 
        client.start(); 
 
        String path = "/app2"; 
 
        TreeCache treeCache = new TreeCache(client, path); 
        treeCache.start(); 
 
        treeCache.getListenable().addListener((client1, event) -> { 
            System.out.println("event.getData()，" + event.getData()); 
            System.out.println("event.getType()，" + event.getType()); 
        }); 
 
        Thread.sleep(Integer.MAX_VALUE); 
    } 
}

當 /app2 的狀態發生變化時，就會調用監聽的方法。

Curator 是對原生的 Zookeeper Api 有封裝的，原生的 Zookeeper 提供的 Api ，注冊監聽后，當數據發生改變時，監聽就被服務端刪除了，要重復注冊監聽。

Curator 則對這個做了相應的封裝和改進。

五、代碼實戰：實現主備選舉

這里我們主要想實現的功能是：

有兩個節點，bigdata001，bigdata002 ，他們互相主備。
bigdata001 啟動時，往 zk 上注冊一個臨時節點 /ElectorLock(鎖)，并且往 /ActiveMaster 下面注冊一個子節點，表示自己是主節點。
bigdata002 啟動時，發現臨時節點 /ElectorLock 存在，表示當前系統已經有主節點了，則自己往 /StandbyMaster 下注冊一個節點，表示自己是 standby。
bigdata001 退出時，釋放 /ElectorLock，并且刪除 /activeMaster 下的節點。
bigdata002 感知到 /ElectorLock 不存在時，則自己去注冊 /ElectorLock，并在 /ActiveMaster 下注冊自己，表示自己已經成為了主節點。

代碼還是用 Curator 框架實現的：

package com.kkarch.zookeeper; 
 
import cn.hutool.core.util.StrUtil; 
import lombok.extern.slf4j.Slf4j; 
import org.apache.curator.framework.CuratorFramework; 
import org.apache.curator.framework.recipes.cache.TreeCache; 
import org.apache.curator.framework.recipes.cache.TreeCacheEvent; 
import org.apache.zookeeper.CreateMode; 
 
import java.nio.charset.StandardCharsets; 
 
/** 
 * 分布式選舉 
 * 
 * @Author wangkai 
 * @Time 2021/7/25 20:12 
 */ 
@Slf4j 
public class ElectorTest { 
 
    private static final String PARENT = "/cluster_ha"; 
    private static final String ACTIVE = PARENT + "/ActiveMaster"; 
    private static final String STANDBY = PARENT + "/StandbyMaster"; 
    private static final String LOCK = PARENT + "/ElectorLock"; 
 
    private static final String HOSTNAME = "bigdata05"; 
    private static final String activeMasterPath = ACTIVE + "/" + HOSTNAME; 
    private static final String standByMasterPath = STANDBY + "/" + HOSTNAME; 
 
    public static void main(String[] args) throws Exception { 
        CuratorFramework zk = ZkUtil.createZkClient("localhost:2181"); 
        zk.start(); 
 
        // 注冊好監聽 
        TreeCache treeCache = new TreeCache(zk, PARENT); 
        treeCache.start(); 
 
        treeCache.getListenable().addListener((client, event) -> { 
            if (event.getType().equals(TreeCacheEvent.Type.INITIALIZED) || event.getType().equals(TreeCacheEvent.Type.CONNECTION_LOST) 
                    || event.getType().equals(TreeCacheEvent.Type.CONNECTION_RECONNECTED) || event.getType().equals(TreeCacheEvent.Type.CONNECTION_SUSPENDED)) { 
                return; 
            } 
            System.out.println(event.getData()); 
            // 如果 Active 下有節點被移除了，沒有節點，則應該去競選成為 Active 
            if (StrUtil.startWith(event.getData().getPath(), ACTIVE) && event.getType().equals(TreeCacheEvent.Type.NODE_REMOVED)) { 
                if (getChildrenNumber(zk, ACTIVE) == 0) { 
                    createZNode(client, LOCK, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL); 
                    System.out.println(HOSTNAME + "爭搶到了鎖"); 
                } 
            } 
            // 如果有鎖節點被創建，則判斷是不是自己創建的，如果是，則切換自己的狀態為 ACTIVE 
            else if (StrUtil.equals(event.getData().getPath(), LOCK) && event.getType().equals(TreeCacheEvent.Type.NODE_ADDED)) { 
                if (StrUtil.equals(new String(event.getData().getData()), HOSTNAME)) { 
                    createZNode(zk, activeMasterPath, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL); 
                    if (checkExists(client, standByMasterPath)) { 
                        deleteZNode(client, standByMasterPath); 
                    } 
                } 
            } 
        }); 
 
        // 先創建 ACTIVE 和 STANDBY 節點 
        if (zk.checkExists().forPath(ACTIVE) == null) { 
            zk.create().creatingParentContainersIfNeeded().forPath(ACTIVE); 
        } 
        if (zk.checkExists().forPath(STANDBY) == null) { 
            zk.create().creatingParentContainersIfNeeded().forPath(STANDBY); 
        } 
 
        // 判斷 ACTIVE 下是否有子節點，如果沒有則去爭搶一把鎖 
        if (getChildrenNumber(zk, ACTIVE) == 0) { 
            createZNode(zk, LOCK, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL); 
        } 
        // 如果有，則自己成為 STANDBY 狀態 
        else { 
            createZNode(zk, standByMasterPath, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL); 
        } 
 
 
        Thread.sleep(1000000000); 
 
 
    } 
 
    public static int getChildrenNumber(CuratorFramework client, String path) throws Exception { 
        return client.getChildren().forPath(path).size(); 
    } 
 
    public static void createZNode(CuratorFramework client, String path, byte[] data, CreateMode mode) { 
        try { 
            client.create().withMode(mode).forPath(path, data); 
        } catch (Exception e) { 
            log.error("創建節點失敗", e); 
            System.out.println("創建節點失敗了"); 
        } 
    } 
 
    public static boolean checkExists(CuratorFramework client, String path) throws Exception { 
        return client.checkExists().forPath(path) != null; 
    } 
 
    public static void deleteZNode(CuratorFramework client, String path) { 
        try { 
            if (checkExists(client, path)) { 
                client.delete().forPath(path); 
            } 
        } catch (Exception e) { 
            log.error("刪除節點失敗", e); 
        } 
    } 
}

責任編輯：武曉燕來源： KK架構

Zookeeper 核心設計