2. 程式人生 > >ZooKeeper的三種典型應用場景

ZooKeeper的三種典型應用場景

阿新 • • 發佈：2019-04-04
sre ddl take consumer ron ica alt needed 開發

引言

　　ZooKeeper是中典型的pub/sub模式的分布式數據管理與協調框架，開發人員可以使用它進行分布式數據的發布與訂閱。另外，其豐富的數據節點類型可以交叉使用，配合Watcher事件通知機制，可以應用於分布式都會涉及的一些核心功能：數據發布/訂閱、Master選舉、命名服務、分布式協調/通知、集群管理、分布式鎖、分布式隊列等。本博文主要介紹：發布/訂閱、分布式鎖、Master選舉三種最常用的場景

　　本文中的代碼示例均是由Curator客戶端編寫的，已經對ZooKeeper原生API做好很多封裝。參考資料《從Paxos到Zookeeper 分布式一致性原理與實踐》（有需要電子PDF的朋友，可以評論私信我）

一、數據發布/訂閱

1、基本概念

（1）數據發布/訂閱系統即所謂的配置中心，也就是發布者將數據發布到ZooKeeper的一個節點或者一系列節點上，提供訂閱者進行數據訂閱，從而實現動態更新數據的目的，實現配置信息的集中式管理和數據的動態更新。ZooKeeper采用的是推拉相結合的方式：客戶端向服務器註冊自己需要關註的節點，一旦該節點的數據發生改變，那麽服務端就會向相應的客戶端發送Wacher事件通知，客戶端接收到消息通知後，需要主動到服務端獲取最新的數據。

（2）實際系統開發過程中：我們可以將初始化配置信息放到節點上集中管理，應用在啟動時都會主動到ZooKeeper服務端進行一次配置讀取，同時在指定節點註冊Watcher監聽，主要配置信息一旦變更，訂閱者就可以獲取讀取最新的配置信息。

通常系統中需要使用一些通用的配置信息，比如機器列表信息、運行時的開關配置、數據庫配置信息等全局配置信息，這些都會有以下3點特性：

　　1） 數據量通常比較小（通常是一些配置文件）

　　2） 數據內容在運行時會經常發生動態變化（比如數據庫的臨時切換等）

　　3） 集群中各機器共享，配置一致（比如數據庫配置共享）。

（3）利用的ZooKeeper特性是：ZooKeeper對任何節點（包括子節點）的變更，只要註冊Wacther事件（使用Curator等客戶端工具已經被封裝好）都可以被其它客戶端監聽

2、代碼示例

技術分享圖片 
package com.lijian.zookeeper.demo;


 
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.util.concurrent.CountDownLatch;

public class ZooKeeper_Subsciption {
    private static final String ADDRESS = "xxx.xxx.xxx.xxx:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static final String PATH = "/configs";
    private static RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    private static String config = "jdbc_configuration";
    private static CountDownLatch countDownLatch = new CountDownLatch(4);

    public static void main(String[] args) throws Exception {
        // 訂閱該配置信息的集群節點（客戶端）:sub1-sub3
        for (int i = 0; i < 3; i++) {
            CuratorFramework consumerClient = getClient();
            subscribe(consumerClient, "sub" + String.valueOf(i));
        }
        // 更改配置信息的集群節點（客戶端）:pub
        CuratorFramework publisherClient = getClient();
        publish(publisherClient, "pub");

    }
    private static void init() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(ADDRESS)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        // 檢查節點是否存在，不存在則初始化創建
        if (client.checkExists().forPath(PATH) == null) {
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(PATH, config.getBytes());
        }
    }


    /**
     * 創建客戶端並且初始化建立一個存儲配置數據的節點
     *
     * @return
     * @throws Exception
     */
    private static CuratorFramework getClient() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(ADDRESS)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        if (client.checkExists().forPath(PATH) == null) {
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(PATH, config.getBytes());
        }
        return client;
    }

    /**
     * 集群中的某個節點機器更改了配置信息：即發布了更新了數據
     *
     * @param client
     * @throws Exception
     */
    private static void publish(CuratorFramework client, String znode) throws Exception {

        System.out.println("節點[" + znode + "]更改了配置數據...");
        client.setData().forPath(PATH, "configuration".getBytes());
        countDownLatch.await();
    }

    /**
     * 集群中訂閱的節點客戶端（機器）獲得最新的配置數據
     *
     * @param client
     * @param znode
     * @throws Exception
     */
    private static void subscribe(CuratorFramework client, String znode) throws Exception {
        // NodeCache監聽ZooKeeper數據節點本身的變化
        final NodeCache cache = new NodeCache(client, PATH);
        // 設置為true：NodeCache在第一次啟動的時候就立刻從ZooKeeper上讀取節點數據並保存到Cache中
        cache.start(true);
        System.out.println("節點["+ znode +"]已訂閱當前配置數據：" + new String(cache.getCurrentData().getData()));
        // 節點監聽
        countDownLatch.countDown();
        cache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() {
                System.out.println("配置數據已發生改變, 節點[" + znode + "]讀取當前新配置數據: " + new String(cache.getCurrentData().getData()));
            }
        });
    }
}
View Code

運行結果：節點[pub]更改了配置數據為“configuration”，訂閱"/configs"節點的sub1-sub3觀測到配置被改變，立馬讀取當前最新的配置數據“configuration”

技術分享圖片

二、Master選舉

1、基本概念

　　（1）在一些讀寫分離的應用場景中，客戶端寫請求往往是由Master處理的，而另一些場景中，Master則常常負責處理一些復雜的邏輯，並將處理結果同步給集群中其它系統單元。比如一個廣告投放系統後臺與ZooKeeper交互，廣告ID通常都是經過一系列海量數據處理中計算得到（非常消耗I/O和CPU資源的過程），那就可以只讓集群中一臺機器處理數據得到計算結果，之後就可以共享給整個集群中的其它所有客戶端機器。

　　（2）利用ZooKeeper的特性：利用ZooKeeper的強一致性，即能夠很好地保證分布式高並發情況下節點的創建一定能夠保證全局唯一性，ZooKeeper將會保證客戶端無法重復創建一個已經存在的數據節點，也就是說如果多個客戶端請求創建同一個節點，那麽最終一定只有一個客戶端請求能夠創建成功，這個客戶端就是Master，而其它客戶端註在該節點上註冊子節點Wacther，用於監控當前Master是否存活，如果當前Master掛了，那麽其余客戶端立馬重新進行Master選舉。

　　（3）競爭成為Master角色之後，創建的子節點都是臨時順序節點，比如：_c_862cf0ce-6712-4aef-a91d-fc4c1044d104-lock-0000000001，並且序號是遞增的。需要註意的是這裏有"lock"單詞，這說明ZooKeeper這一特性，也可以運用於分布式鎖。

　　　技術分享圖片

2、代碼示例

技術分享圖片 
package com.lijian.zookeeper.demo;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class ZooKeeper_Master {

    private static final String ADDRESS="xxx.xxx.xxx.xxx:2181";
    private static final int SESSION_TIMEOUT=5000;
    private static final String MASTER_PATH = "/master_path";
    private static final int CLIENT_COUNT = 5;

    private static RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);


    public static void main(String[] args) throws InterruptedException {

        ExecutorService service = Executors.newFixedThreadPool(CLIENT_COUNT);
        for (int i = 0; i < CLIENT_COUNT; i++) {
            final String index = String.valueOf(i);
            service.submit(() -> {
                masterSelect(index);
            });
        }
    }

    private static void  masterSelect(final String znode){
        // client成為master的次數統計
        AtomicInteger leaderCount = new AtomicInteger(1);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(ADDRESS)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        // 一旦執行完takeLeadership，就會重新進行選舉
        LeaderSelector selector = new LeaderSelector(client, MASTER_PATH, new LeaderSelectorListenerAdapter() {
            @Override
            public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
                System.out.println("節點["+ znode +"]成為master");
                System.out.println("節點["+ znode +"]已經成為master次數："+ leaderCount.getAndIncrement());
                // 睡眠5s模擬成為master後完成任務
                Thread.sleep(5000);
                System.out.println("節點["+ znode +"]釋放master");
            }
        });
        // autoRequeue自動重新排隊：使得上一次選舉為master的節點還有可能再次成為master
        selector.autoRequeue();
        selector.start();
    }
}
View Code

運行結果：由於執行selector.autoRequeue()方法，被選舉為master後的節點可能會再次獲被選舉為master，所以會一直循環執行，以下只截圖部分。其中獲取成為master的次數充分表明了Master選舉的公平性。

　技術分享圖片

三、分布式鎖

1、基本概念

　　（1）對於排他鎖：ZooKeeper通過數據節點表示一個鎖，例如/exclusive_lock/lock節點就可以定義一個鎖，所有客戶端都會調用create()接口，試圖在/exclusive_lock下創建lock子節點，但是ZooKeeper的強一致性會保證所有客戶端最終只有一個客戶創建成功。也就可以認為獲得了鎖，其它線程Watcher監聽子節點變化（等待釋放鎖，競爭獲取資源）。

　　　　 對於共享鎖：ZooKeeper同樣可以通過數據節點表示一個鎖，類似於/shared_lock/[Hostname]-請求類型（讀/寫）-序號的臨時節點，比如/shared_lock/192.168.0.1-R-0000000000

2、代碼示例

Curator提供的有四種鎖，分別如下：

　　（1）InterProcessMutex：分布式可重入排它鎖

　　（2）InterProcessSemaphoreMutex：分布式排它鎖

　　（3）InterProcessReadWriteLock：分布式讀寫鎖

　　（4）InterProcessMultiLock：將多個鎖作為單個實體管理的容器

主要是以InterProcessMutex為例，編寫示例：

技術分享圖片 
package com.lijian.zookeeper.demo;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ZooKeeper_Lock {
    private static final String ADDRESS = "xxx.xxx.xxx.xxx:2181";
    private static final int SESSION_TIMEOUT = 5000;
    private static final String LOCK_PATH = "/lock_path";
    private static final int CLIENT_COUNT = 10;

    private static RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
    private static int resource = 0;

    public static void main(String[] args){
        ExecutorService service = Executors.newFixedThreadPool(CLIENT_COUNT);
        for (int i = 0; i < CLIENT_COUNT; i++) {
            final String index = String.valueOf(i);
            service.submit(() -> {
                distributedLock(index);
            });
        }
    }

    private static void distributedLock(final String znode) {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(ADDRESS)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        final InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH);
        try {
//            lock.acquire();
            System.out.println("客戶端節點[" + znode + "]獲取lock");
            System.out.println("客戶端節點[" + znode + "]讀取的資源為：" + String.valueOf(resource));
            resource ++;
//            lock.release();
            System.out.println("客戶端節點[" + znode + "]釋放lock");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
View Code

運行結果：加鎖後可以從左圖看到讀取的都是最新的資源值。如果去掉鎖的話讀取的資源值不能保證是最新值看右圖

技術分享圖片　　　　技術分享圖片

ZooKeeper的三種典型應用場景

相關推薦

ZooKeeper的三種典型應用場景

sre ddl take consumer ron ica alt needed 開發 引言 　　ZooKeeper是中典型的pub/sub模式的分布式數據管理與協調框架，開發人員可以使用它進行分布式數據的發布與訂閱。另外，其豐富的數據節點類型可以交叉使用，配合Wat

ZooKeeper的典型應用場景

拉取 ons 執行 全局 進行 創建失敗 消息通知 防止 成了 《從Paxos到Zookeeper 分布式一致性原理與實踐》讀書筆記 本文：總結腦圖地址：腦圖 前言 所有的典型應用場景，都是利用了ZK的如下特性： 強一致性：在高並發情況下，能夠保證節點的創建一定是

SpringBoot整合RabbitMQ之典型應用場景實戰三

分布 boot 自動刪除 blog jce 地址 這樣的 實施 微服務 實戰前言RabbitMQ 作為目前應用相當廣泛的消息中間件，在企業級應用、微服務應用中充當著重要的角色。特別是在一些典型的應用場景以及業務模塊中具有重要的作用，比如業務服務模塊解耦、異步通信、高並發限流

ZooKeeper 典型應用場景

Zookeeper基礎知識　　1.zookeeper是一個類似hdfs的樹形檔案結構，zookeeper可以用來保證資料在(zk)叢集之間的資料的事務性一致、　　2.zookeeper有watch事件，是一次性觸發的，當watch監視的資料發生變化時，通知設定了該watch的client，即watcher　　

ZooKeeper典型應用場景

ZooKeeper是一個高可用的分散式資料管理與系統協調框架。基於對Paxos演算法的實現，使該框架保證了分散式環境中資料的強一致性，也正是基於這樣的特性，使得ZooKeeper解決很多分散式問題。網上對ZK的應用場景也有不少介紹，本文將結合作者身邊的專案例子，系統地對ZK的

ZooKeeper典型應用場景一覽

叢集機器監控：這通常用於那種對叢集中機器狀態，機器線上率有較高要求的場景，能夠快速對叢集中機器變化作出響應。這樣的場景中，往往有一個監控系統，實時檢測叢集機器是否存活。過去的做法通常是：監控系統通過某種手段（比如ping）定時檢測每個機器，或者每個機器自己定時向監控系統彙報“我還活著”。 這種做法可行

zookeeper開源客戶端Curator典型應用場景之-服務註冊與發現(十一)

隨著業務增加,以前簡單的系統已經變得越來越複雜，單純的提升伺服器效能也不是辦法,而且程式碼也是越來越龐大，維護也變得越來越困難，這一切都催生了新的架構設計風格 – 微服務架構的出現。 微服務給我們帶來了很多好處，例如：獨立可擴充套件、易維護。但是隨著應用的分解

zookeeper開源客戶端Curator典型應用場景之-Barrier屏障(十三)

什麼是Barrier Barrier是這樣的：Barrier是一個同步點，每一個程序到達此點都要等待，直到某一個條件滿足，然後所有的節點繼續進行。 比如：賽跑大家都知道,所有比賽人員都會在起跑線外等待,直到教練員的槍響之後,所有參賽者立刻開始賽跑。 JDK的併

圖解ZooKeeper的典型應用場景

介紹 zookeeper在很多框架中都有應用，例如：Dubbo，Hadoop，Storm，Kafka等，在這些框架中都用到了zookeeper，但典型的用法也就幾種，掌握了這幾種用法，再看zookeeper在相關框架中的應用就很輕鬆，下一篇文章將會詳細介紹zookeeper在dubb

大資料生態之zookeeper（典型應用場景）

1. 命名服務        命名服務是分散式系統中較為常見的一類場景，分散式系統中，被命名的實體通常可以是叢集中的機器、提供的服務地址或者遠端物件，通過命名服務，客戶端可以根據指定名字來獲取資源的實體、服務地址和提供者的資訊。Zookee

【推薦】zookeeper典型應用場景： 分散式計數器

一、技術介紹 zookeeper有很多典型應用場景，應用在分散式系統中，這裡介紹其分散式計數器應用。本文將討論如何使用Curator來實現計數器。 顧名思義，計數器是用來計數的, 利用ZooKeeper可以實現一個叢集共享的計數器。 只要使用相同的path就可以得到最新的計數器值， 這是由Z

zookeeper開源客戶端Curator典型應用場景之-分散式計數器(十四)

之前我們瞭解了基於Corator的分散式鎖之後,我們就很容易基於其實現一個分散式計數器，顧名思義，計數器是用來計數的, 利用ZooKeeper可以實現一個叢集共享的計數器。 只要使用相同的path就可以得到最新的計數器值， 這是由ZooKeeper的一致性保證

ZooKeeper的典型應用場景之Master選舉。

        Master選舉是一個分散式系統中非常常見的應用場景。分散式最核心的特性就是能夠將具有獨立計算能力的系統單元部署在不同的機器上，構成一個完整的分散式系統。而與此同時，實際場景中往往也需要在這些分佈在不同機器上的獨立系統單元中選出一個所謂的“老大”，在電腦科學

ZooKeeper的典型應用場景之名稱空間。

        命名服務（Name Service）也是分散式系統中比較常見的一類場景，在《Java網路高階程式設計》一書中提到，命名服務是分散式系統最基本的公共服務之一。在分散式系統中，被命名的實體通常可以是叢集中的機器、提供的服務地址或遠端物件等——這些我們都可以統稱他

zookeeper開源客戶端Curator典型應用場景之-訊息佇列(十二)

Curator框架也有分散式佇列實現。 利用ZK的PERSISTENT SEQUENTIAL(持久順序)節點，可以保證放入到佇列中的專案是按照順序排隊的。並且宕機重啟並不丟失訊息， 如果單一的消費者從佇列中取資料， 那麼它是先入先出的，這也是佇列的特點。 如果

zookeeper開源客戶端Curator典型應用場景之-Master選舉(十)

在生產環境中，一般要保證服務的高可用，有時候只需要選出一臺機器來執行，其餘機器處於備用狀態，比如，在分散式系統中很常見的一個問題就是定時任務的執行。如果多臺機器同時執行相同的定時任務，業務複雜則可能出現災難性的後果。我使用的是噹噹網的elastic-job分散

ZooKeeper學習之路 （七）ZooKeeper設計特點及典型應用場景

目錄 正文 回到頂部 ZooKeeper 特點/設計目的 ZooKeeper 作為一個叢集提供資料一致的協調服務，自然，最好的方式就是在整個叢集中的 各服務節點進行資料的複製和同步。 資料複製的好處 1、容錯：一個節點出錯，不至於讓整個叢集無法提供服務

ZooKeeper設計特點及典型應用場景

資料複製的好處 1、容錯：一個節點出錯，不至於讓整個叢集無法提供服務 2、擴充套件性：通過增加伺服器節點能提高 ZooKeeper 系統的負載能力，把負載分佈到多個節點上 3、高效能：客戶端可訪問本地 ZooKeeper 節點或者訪問就近的節點，依次提高使用者的訪問速度 設計目的 1、最終一

淺析DES與AES、RSA三種典型加密算法的比較

問題 通訊 無法 尋找 rsa算法 結合 rsa加密 軟件技術 -c DES與AES的比較 自DES 算法公諸於世以來，學術界圍繞它的安全性等方面進行了研究並展開了激烈的爭論。在技術上，對DES的批評主要集中在以下幾個方面： 1、作為分組密碼，DES 的加密單位僅有64

幾種典型應用對系統資源使用的特點

java 訪問量 資源 虛擬內存 數據庫應用 1）以靜態內容為主的Web應用a、小文件居多，並且讀操作頻繁，主要消耗內容 b、優化，盡量在前段使用緩存，如果訪問量較大的情況，使用cdn c、在高並發訪問時，還存在另外一個，如果網絡帶寬瓶頸，客戶訪問量大且帶寬不夠，就會阻塞網絡2）以動態內容