[[413943]]

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

一、前言

想起很久以前在某個客戶現(xiàn)場，微服務 B 突然無法調(diào)用到微服務 A，為了使服務盡快正?；謴?，重啟了微服務 B 。

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

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

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

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

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

二、Zookeeper 核心架構設計

1、Zookeeper 特點

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

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

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

其中一個節(jié)點作為主節(jié)點，提供分布式事務的寫服務，其他節(jié)點和這個節(jié)點同步數(shù)據(jù)，保持和主節(jié)點狀態(tài)一致。

(3)Zookeeper 所有節(jié)點的數(shù)據(jù)狀態(tài)通過 Zab 協(xié)議保持一致。當集群中沒有 Leader 節(jié)點時，內(nèi)部會執(zhí)行選舉，選舉結束，F(xiàn)ollower 和 Leader 執(zhí)行狀態(tài)同步;當有 Leader 節(jié)點時，Leader 通過 ZAB 協(xié)議主導分布式事務的執(zhí)行，并且所有的事務都是串行執(zhí)行的。

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

2、Zookeeper 角色的理解

Zookeeper 并沒有沿用 Master/Slave 概念，而是引入了 Leader，F(xiàn)ollower，Observer 三種角色。

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

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

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

3、Zookeeper 同時滿足了 CAP 嗎?

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

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

也就是說，Zookeeper 并不保證讀取的一定是最新的數(shù)據(jù)。如果一定要最新，需要使用 sync 回調(diào)處理。

三、核心機制一：ZNode 數(shù)據(jù)模型

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

1、ZNode 并不適合存儲太大的數(shù)據(jù)

為什么是類文件系統(tǒng)呢?因為 ZNode 模型沒有文件和文件夾的概念，每個節(jié)點既可以有子節(jié)點，也可以存儲數(shù)據(jù)。

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

原因有以下四點：

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

2、ZNode 的分類

(1)按生命周期分類

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

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

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

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

不帶序號，就是創(chuàng)建什么就是什么

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

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

(4)注意的點

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

3. 代碼實戰(zhàn)

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

這里我們使用 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; 
 
    // 連接超時 、 有啥區(qū)別 
    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 { 
        // 創(chuàng)建客戶端 
        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()); 
 
        // 創(chuàng)建節(jié)點 /test1 
        client.create() 
                .forPath("/test1", "curator data".getBytes(StandardCharsets.UTF_8)); 
 
        System.out.println(client.getChildren().forPath("/")); 
 
        // 臨時節(jié)點 
        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)); 
        // 遞歸創(chuàng)建 
        client.create() 
                .creatingParentContainersIfNeeded() 
                .forPath("/secondPath1/sencond2/sencond3"); 
 
 
        Thread.sleep(10000); 
    } 

四、核心機制二：Watcher 監(jiān)聽機制

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

1. 對于 Watcher 機制的理解

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

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

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

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

在這個場景中，老板是發(fā)布者，員工是訂閱者，釘釘群就是 Zookeeper 系統(tǒng)。

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

2、 Watcher 機制的流程

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

監(jiān)聽器 Watcher 的定義：

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

下面是監(jiān)聽的大致流程圖：

稍稍解釋一下：

1、Client1 和 Client2 都關心 /app2 節(jié)點的數(shù)據(jù)狀態(tài)變化，于是注冊一個對于 /app2 的監(jiān)聽器到 Zookeeper 上;

2、當 Client3 修改 /app2 的值后，Zookeeper 會主動通知 Client1 和 Client2 ，并且回調(diào)監(jiān)聽器的方法。

當然這里的數(shù)據(jù)狀態(tài)變化有下面這些類型：

• 節(jié)點被創(chuàng)建;
• 節(jié)點被刪除;
• 節(jié)點數(shù)據(jù)發(fā)生改變;
• 節(jié)點的子節(jié)點個數(shù)發(fā)生改變。

3. 通過代碼來初步理解

我們還是用 Curator 框架來驗證一下這個監(jiān)聽器。

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

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 的狀態(tài)發(fā)生變化時，就會調(diào)用監(jiān)聽的方法。

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

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

五、代碼實戰(zhàn)：實現(xiàn)主備選舉

這里我們主要想實現(xiàn)的功能是：

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

代碼還是用 Curator 框架實現(xiàn)的：

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(); 
 
        // 注冊好監(jiān)聽 
        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 下有節(jié)點被移除了，沒有節(jié)點，則應該去競選成為 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 + "爭搶到了鎖"); 
                } 
            } 
            // 如果有鎖節(jié)點被創(chuàng)建，則判斷是不是自己創(chuàng)建的，如果是，則切換自己的狀態(tài)為 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); 
                    } 
                } 
            } 
        }); 
 
        // 先創(chuàng)建 ACTIVE 和 STANDBY 節(jié)點 
        if (zk.checkExists().forPath(ACTIVE) == null) { 
            zk.create().creatingParentContainersIfNeeded().forPath(ACTIVE); 
        } 
        if (zk.checkExists().forPath(STANDBY) == null) { 
            zk.create().creatingParentContainersIfNeeded().forPath(STANDBY); 
        } 
 
        // 判斷 ACTIVE 下是否有子節(jié)點，如果沒有則去爭搶一把鎖 
        if (getChildrenNumber(zk, ACTIVE) == 0) { 
            createZNode(zk, LOCK, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL); 
        } 
        // 如果有，則自己成為 STANDBY 狀態(tài) 
        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("創(chuàng)建節(jié)點失敗", e); 
            System.out.println("創(chuàng)建節(jié)點失敗了"); 
        } 
    } 
 
    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("刪除節(jié)點失敗", e); 
        } 
    } 
}