zookeeper原生客戶端

阿新 • • 發佈：2018-11-08

一：zookeeper常用客戶端

zookeeper：官方提供的，原生的api，使用起來比較麻煩，比較底層，不夠直接，不建議使用。
zkclient: 對原生api的封裝，開源專案(https://github.com/adyliu/zkclient)，dubbo中使用的是這個。
Apache Curator：Apache的開源專案，對原生客戶端zookeeper進行封裝，易於使用, 功能強大，一般都是使用這個框架。

二：zookeeper原生客戶端

<dependency>
    <groupId>org.apache.zookeeper</groupId 
>
    <artifactId>zookeeper</artifactId>
    <version>3.4.12</version>
</dependency>

public enum CreateMode {
     /** 持久節點 */
    PERSISTENT(0, false, false),
    /** 持久順序節點 */
    PERSISTENT_SEQUENTIAL(2, false, true),
    /** 臨時節點(本次會話有效，會話結束後會自動刪除) */
    EPHEMERAL(1, true 
, false),
    /** 臨時順序節點 */
    EPHEMERAL_SEQUENTIAL(3, true, true);
}

public class ZooKeeper {
    public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException;

    // 不存在返回null
    public Stat exists(String path, boolean watch) throws KeeperException, InterruptedException;

    /**
    * 重複建立同一個節點會拋異常
    * 建立子節點必選先保證父節點已經建立好了
    */ 

    public String create(String path, byte[] data, List<ACL> acl, CreateMode createMode) throws KeeperException, InterruptedException;

    public byte[] getData(String path, boolean watch, Stat stat) throws KeeperException, InterruptedException;

    public List<String> getChildren(String path, boolean watch) throws KeeperException, InterruptedException;

    public Stat setData(String path, byte[] data, int version) throws KeeperException, InterruptedException;

    /**
    * 只能刪除一個節點，不支援遞迴的刪除某個路徑
    * 刪除的節點不能包含子節點
    */
    public void delete(String path, int version) throws InterruptedException, KeeperException;

    // 新增認證資訊(類似於密碼)
    public void addAuthInfo(String scheme, byte[] auth);
    // ACL(Access Control List)設定節點訪問許可權列表，每個節點都可以設定訪問許可權，指定只有特定的客戶端才能訪問和操作節點。
    public Stat setACL(String path, List<ACL> acl, int aclVersion) throws KeeperException, InterruptedException;
    public List<ACL> getACL(String path, Stat stat) throws KeeperException, InterruptedException;

    public synchronized void close() throws InterruptedException;
}

public class ZooDefs {
    public interface Ids {
        Id ANYONE_ID_UNSAFE = new Id("world", "anyone");
        Id AUTH_IDS = new Id("auth", "");

        /** 這是一個完全開放的許可權，所有客戶端都有許可權 */
        ArrayList<ACL> OPEN_ACL_UNSAFE = new ArrayList(Collections.singletonList(new ACL(31, ANYONE_ID_UNSAFE)));
        /** 只有建立節點的客戶端才有所有許可權 */
        ArrayList<ACL> CREATOR_ALL_ACL = new ArrayList(Collections.singletonList(new ACL(31, AUTH_IDS)));
        /** 所有客戶端只有讀取的 */
        ArrayList<ACL> READ_ACL_UNSAFE = new ArrayList(Collections.singletonList(new ACL(1, ANYONE_ID_UNSAFE)));
    }
}

EventType

public static enum EventType {
    None(-1),
    NodeCreated(1),
    NodeDeleted(2),
    NodeDataChanged(3),
    NodeChildrenChanged(4);
}

KeeperState

public static enum KeeperState {
    Disconnected(0),
    SyncConnected(3),
    AuthFailed(4),
    ConnectedReadOnly(5),
    SaslAuthenticated(6),
    Expired(-112);
}

注意：程式不能debug，只能run，如果debug會報錯org.apache.zookeeper.KeeperException$SessionExpiredException: KeeperErrorCode = Session expired for /app1
原因是debug造成程式停止執行，導致會話過期

public class ZookeeperTest {
    /** zookeeper地址 */
    static final String connectString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
    /** session超時時間 */
    static final int sessionTimeout = 5000;
    /** 阻塞程式執行，用於等待zookeeper連線成功，傳送成功訊號 */
    static final CountDownLatch countDownLatch = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
        Watcher watcher = new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // 事件狀態
                Event.KeeperState keeperState = watchedEvent.getState();
                if (Event.KeeperState.SyncConnected == keeperState) {
                    // 事件型別
                    Event.EventType eventType = watchedEvent.getType();
                    if (Event.EventType.None == eventType) {
                        // 連線成功建立，傳送訊號量，讓後續阻塞程式向下執行
                        countDownLatch.countDown();
                        System.out.println("連線成功：" + watchedEvent);
                    }
                }
            }
        };
        // 注意：ZooKeeper客戶端和伺服器會話的建立是一個非同步的過程，也就是說程式方法在處理完客戶端初始化後立即返回
        // 也就是說可能並沒有真正構建好一個可用的會話，只有會話宣告週期處於SyncConnected時才算真正建立好會話
        // 這也是為什麼要使用CountDownLatch來等待連線成功
        ZooKeeper zooKeeper = new ZooKeeper(connectString, sessionTimeout, watcher);
        // 新增認證資訊
        // digest:是最常見的許可權控制模式，也更符合我們對許可權控制的認證，類似於使用者名稱和密碼的方式，zk會對形成的許可權標識先後進行兩次編碼處理，分別是SHA-1加密演算法、Base64編碼
        zooKeeper.addAuthInfo("digest", "123456".getBytes());
        // 如果沒有連線成功則進行阻塞，直到連線成功才繼續往下執行，連線時需要時間的，可能不會立即連線成功，肯能會等一兩秒之後才連線成功，
        countDownLatch.await();


        String app1Node = "/app1";
        try {
            Stat app1Stat = zooKeeper.exists(app1Node, false);
            if (app1Stat == null) {
                List<ACL> acls = new ArrayList<>(1);
                for (ACL acl : ZooDefs.Ids.CREATOR_ALL_ACL) {
                    acls.add(acl);
                }

                // 建立節點(如果節點已存在則丟擲異常)，返回節點名稱
                // 該節點需要認證，只有認證的其它zk客戶端才能操作該節點
                String app1NodePath = zooKeeper.create(app1Node, "app1".getBytes(), acls, CreateMode.PERSISTENT);
                System.out.println("app1NodePath=" + app1NodePath);

                Stat p1Stat = zooKeeper.exists(app1Node + "/p_1", false);
                if (p1Stat == null) {
                    // 建立子節點(父節點必須存在，父節點不存在則拋異常)
                    // CreateMode.EPHEMERAL 表示臨時節點，當會話結束後會立即刪除
                    zooKeeper.create(app1Node + "/p_1", "p_1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                    zooKeeper.create(app1Node + "/p_2", "p_2".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                    zooKeeper.create(app1Node + "/p_3", "p_3".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                }
            }


            byte[] data = zooKeeper.getData(app1Node, false, null);
            String app1NodeValue = new String(data);
            System.out.println("app1NodeValue=" + app1NodeValue);

            zooKeeper.setData(app1Node, "my p_1".getBytes(), -1);

            List<String> children = zooKeeper.getChildren(app1Node, false);
            for (String path : children) {
                String fullPath = app1Node + "/" + path;
                String nodeValue = new String(zooKeeper.getData(fullPath, false, null));
                System.out.println(fullPath + "=" + nodeValue);
            }

            // version版本號 -1: 全刪除
            zooKeeper.delete(app1Node + "/p_2", -1);

            // Thread.sleep(10000);
        } catch (Exception e) {
            System.out.println(e);
        } finally {
            zooKeeper.close();
        }
    }
}

這裡寫圖片描述

原生zookeeper客戶端使用注意：

ZooKeeper建立連線時是非同步建立的，有可能已經開始呼叫客戶端方法了，連線還沒有完全建立好，所以在建立連線時一般將非同步建立客戶端變成同步建立客戶端
session過期的問題: 在極端情況下,出現ZooKeeper session過期,客戶端需要自己去監聽該狀態並重新建立ZooKeeper例項
自動恢復(failover)的問題: 當client與一臺server的連線丟失,並試圖去連線另外一臺server時, client將回到初始連線模式
對可恢復異常的處理:當在server端建立一個有序ZNode,而在將節點名返回給客戶端時崩潰,此時client端丟擲可恢復的異常,使用者需要自己捕獲這些異常並進行重試
使用場景的問題:Zookeeper提供了一些標準的使用場景支援,但是ZooKeeper對這些功能的使用說明文件很少,而且很容易用錯.在一些極端場景下如何處理,zk並沒有給出詳細的文件說明.比如共享鎖服務,當伺服器端建立臨時順序節點成功,但是在客戶端接收到節點名之前掛掉了,如果不能很好的處理這種情況,將導致死鎖

三：Watcher示例

Watcher：監控者，監控某個節點對應的操作，也就是對某個節點操作時執行一下回調函式，Watcher#process()

如果某個動作(操作節點)需要進行監控需要設定某個動作的watch=true或者設定exists(path, true)。注意執行一次動作和一次watch繫結，如果想每次都需要監控，那麼每次動作都必須線上設定監控

public class ZookeeperWatcher implements Watcher {
    /** zookeeper地址 */
    private static final String connectString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
    /** session超時時間 */
    private static final int sessionTimeout = 5000;
    /** 阻塞程式執行，用於等待zookeeper連線成功，傳送成功訊號 */
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);

    private AtomicInteger seq = new AtomicInteger();

    private ZooKeeper zooKeeper;

    private static final String PARENT_PATH = "/p";
    private static final String CHILDEREN_PATH = "/p/c";

    public void createConnection(String connectString, int sessionTimeout){
        try {
            this.releaseConnection();
            zooKeeper = new ZooKeeper(connectString, sessionTimeout, this);
            countDownLatch.await();
        } catch (Exception e) {

        }
    }

    public void releaseConnection() {
        if (this.zooKeeper == null) return;

        try {
            this.zooKeeper.close();
            System.out.println("斷開連線");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public boolean createPath(String path, String data, boolean needWatch) {
        try {
            // 設定監控(由於zookeeper的監控都是一次性的，所以每次都需要重新設定監控)
            System.out.println("新增節點");
            this.zooKeeper.exists(path, needWatch);
            this.zooKeeper.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("[" + Thread.currentThread().getName() +  "] 節點建立成功 path=" + path + ", content=" + data);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

        return true;
    }

    public void writeData(String path, String data) {
        System.out.println("修改節點");
        try {
            this.zooKeeper.setData(path, data.getBytes(), -1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public List<String> getChildren(String path, boolean needWatch) {
        try {
            System.out.println("獲取子節點");
            return this.zooKeeper.getChildren(path, needWatch);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    public void deleteAllTestPath(boolean needWatch) {
        if (this.reSetWatch(CHILDEREN_PATH) != null) {
            this.deleteNode(CHILDEREN_PATH);
        }

        if (this.reSetWatch(PARENT_PATH) != null) {
            this.deleteNode(PARENT_PATH);
        }
    }

    public void deleteNode(String path) {
        try {
            System.out.println("刪除節點：" + path);
            this.zooKeeper.delete(path, -1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public Stat reSetWatch(String path){
        try {
            // 重新設定監控
            return this.zooKeeper.exists(path, true);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    @Override
    public void process(WatchedEvent watchedEvent) {
        System.out.println("process watchedEvent=" + watchedEvent);
        if (watchedEvent == null) return;
        Event.KeeperState keeperState = watchedEvent.getState();
        Event.EventType eventType = watchedEvent.getType();
        String path = watchedEvent.getPath();

        String prefix = "[Watcher(" + Thread.currentThread().getName() + ") -" + this.seq.incrementAndGet() + "] path=" + path + "\t";
        System.out.println(prefix + "連線狀態：" + keeperState);
        System.out.println(prefix + "事件：" + eventType);

        if (Event.KeeperState.SyncConnected == keeperState) {
            if (Event.EventType.None == eventType) {
                System.out.println(prefix + "成功連線上ZK伺服器 》》》》》");
                countDownLatch.countDown();
            } else if (Event.EventType.NodeCreated == eventType) {
                System.out.println(prefix + "建立節點");
                this.sleep(100);
            } else if (Event.EventType.NodeDataChanged == eventType) {
                System.out.println(prefix + " 修改節點");
                this.sleep(100);
            } else if (Event.EventType.NodeChildrenChanged == eventType) {
                System.out.println(prefix + " 修改子節點");
                this.sleep(3000);
            } else if (Event.EventType.NodeDeleted == eventType) {
                System.out.println(prefix + " 刪除子節點 " + path);
                this.sleep(3000);
            }
        } else if (Event.KeeperState.Disconnected == keeperState) {
            System.out.println(prefix + " 連線斷開 XXXXX");
        } else if (Event.KeeperState.AuthFailed == keeperState) {
            System.out.println(prefix + " 許可權檢查失敗");
        } else if (Event.KeeperState.Expired == keeperState) {
            System.out.println(prefix + " 會話過期");
        }

        System.out.println("--------------------------------------------------------------------");
    }

    private void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        ZookeeperWatcher zkWatcher = new ZookeeperWatcher();
        // 第一次觸發watcher 建立連線會觸發
        zkWatcher.createConnection(connectString, sessionTimeout);
        Thread.sleep(1000);

        // true：表示需要監控, 會觸發第二次watcher
        if (zkWatcher.createPath(PARENT_PATH, System.currentTimeMillis() + "", true)) {
            Thread.sleep(1000);

            // 重新建立監控(每次監控完就會斷開，下次要想還監控必須手動再次設定監控)
            zkWatcher.reSetWatch(PARENT_PATH);

            // 第三次觸發監控(因監控是一次性的，createPath監控用完了就斷開監控了，而上面又再次建立了監控，所以writeData能被監控到)
            zkWatcher.writeData(PARENT_PATH, System.currentTimeMillis() + "");

            // 監控父節點(當下面的程式碼執行時，父節點增加了一個子節點也屬於父節點發生了變化)
            zkWatcher.getChildren(PARENT_PATH, true);

            // 監控子節點，會觸發監控
            zkWatcher.createPath(CHILDEREN_PATH, System.currentTimeMillis() + "", true);

            // 觸發兩次監控
            zkWatcher.deleteAllTestPath(true);

        }

        zkWatcher.releaseConnection();

        Thread.sleep(10000);
    }
}

這裡寫圖片描述

zookeeper原生客戶端

一：zookeeper常用客戶端 zookeeper：官方提供的，原生的api，使用起來比較麻煩，比較底層，不夠直接，不建議使用。 zkclient: 對原生api的封裝，開源專案(https://github.com/adyliu/zkclient)，dubbo中使用

Zookeeper之Zookeeper底層客戶端架構實現原理（轉載）

一次描述綁定機制一個 ini fin 源碼 receive Zookeeper的Client直接與用戶打交道，是我們使用Zookeeper的interface。了解ZK Client的結構和工作原理有利於我們合理的使用ZK，並能在使用中更早的發現問題。本文將在研究源

zookeeper的客戶端應用

什麼zookeeper? 　　ZooKeeper是一個分散式的，開放原始碼的分散式應用程式協調服務，是Google的Chubby一個開源的實現，是Hadoop和Hbase的重要元件。它是一個為分散式應用提供一致性服務的軟體，提供的功能包括：配置維護、域名服務、分散式同步、組服務等。 ZooKeeper

zookeeper之客戶端和服務端的區別

客戶端是叢集外的訪問，服務端才是叢集上的提供服務的。使用bin/zkServer.sh start開啟的zookeeper上的一個服務端，而使用bin/zkCli.sh是將客戶端連到服務端上。客戶端可以通過服務端建立znode,刪除znode，寫znode,讀znod

zookeeper(三)——客戶端指令碼

zkCli.sh介紹：進入bin目錄下執行命令： sh zkCli.sh 當看到如下輸出資訊時，表示連線成功上本地的zookeeper伺服器了： WatchedEvent state:SyncConnected type:None path:null [zk:

zookeeper開源客戶端Curator介紹(六)

上一篇文章介紹了zookeeper原生API的使用,使用過原生API不得不說,有很多的問題,比如：不能遞迴建立和刪除節點、Watcher只能使用一次、還有很多可以解決分散式應用問題的api(比如分散式鎖,leader選舉等)，但由於ZooKeeper提供的原始

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

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

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

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

zookeeper簡單客戶端API

public class ZKclient{ public static void main(String[] args){ private String connectString="ip1:2181,ip2:2181,ip3:2181"; private int sessio

zookeeper(5)客戶端

　　zookeeper客戶端主要負責與使用者進行互動，將命令傳送到伺服器，接收伺服器的響應，反饋給使用者。主要分為一下三層：使用者命令處理層　　使用者命令處理層的功能是讀取使用者輸入的命令，解析使用者命令和輸入引數，根據命令和引數，進行一些校驗，然後執行節點操作。原始碼例項（Zoo

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

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

Zookeeper C客戶端解析

Zookeeper客戶端主要有以下幾個核心元件組成： Zookeeper控制代碼：儲存預設Watcher、Zookeeper伺服器地址列表等； Adaptor：包含兩個執行緒：I/O執行緒和Completion執行緒，其中，I/O執行緒負責客戶端與服務端之間的I/O通訊；Comp

zookeeper java 客戶端ACL許可權使用

zookeeper 提供許可權認證作為zookeeper客戶端訪問的限制，主要有兩種方式，1、IP模式 2、 digest許可權模式可以通過建立節點時定義許可權內容。以下是java的實現 package com.aicong.test.helloZookeeper;

Zookeeper 開源客戶端 ZkClient 版本 api介紹和示例

ZkClient是由Datameer的工程師開發的開源客戶端，對Zookeeper的原生API進行了包裝，實現了超時重連、Watcher反覆註冊等功能。 ZKClient版本及原始碼 maven依賴 ZkClient目前有兩個不同artifactId的系列。其中最早

Zookeeper開源客戶端框架Curator簡介

curator簡介 Netflix curator 是Netflix公司開源的一個Zookeeper client library，用於簡化zookeeper客戶端程式設計，包含一下幾個模組： curator-client - zookeeper client封裝，用於

Zookeeper開源客戶端Curator之基本功能講解

簡介 Curator是Netflix公司開源的一套Zookeeper客戶端框架。瞭解過Zookeeper原生API都會清楚其複雜度。Curator幫助我們在其基礎上進行封裝、實現一些開發細節，包括接連重連、反覆註冊Watcher和NodeExistsExcept

ZooKeeper——Curator客戶端基本使用（筆記）

Curator使用 Curator是Netflix公司一個開源的zookeeper客戶端，在原生API介面上進行了包裝，解決了很多ZooKeeper客戶端非常底層的細節開發。同時內部實現了諸如Session超時重連，Watcher反覆註冊等功能，實現了Flue

關於zookeeper第三方客戶端zkclient的使用說明

ZkClient 在使用ZooKeeper的Java客戶端時，經常需要處理幾個問題：重複註冊watcher、session失效重連、異常處理。要解決上述的幾個問題，可以自己解決，也可以採用第三方的java客戶端來完成。這裡就介

【Zookeeper】——客戶端簡介

本篇部落格介紹客戶端，客戶端是開發人員使用Zookeeper最主要的途徑，包括Zookeeper的客戶端和第三方提供的客戶端。 Zookeeper的客戶端對Zookeeper的客戶端

【hadoop zookeeper】Zookeeper開源客戶端框架Curator簡介

Curator是Netflix開源的一套ZooKeeper客戶端框架. Netflix在使用ZooKeeper的過程中發現ZooKeeper自帶的客戶端太底層, 應用方在使用的時候需要自己處理很多事情, 於是在它的基礎上包裝了一下, 提供了一套更好用的客戶端框架. Netf

zookeeper原生客戶端

一：zookeeper常用客戶端

二：zookeeper原生客戶端

三：Watcher示例

相關推薦