[[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 系統。

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

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); 
        } 
    } 
}