碼上快樂

相關內容    簡體    繁體

Apache Zookeeper Java客戶端Curator使用及權限模式詳解

本文轉載自   查看原文   2021-10-25 09:43   172 

這篇文章是讓大家了解Zookeeper基於Java客戶端Curator的基本操作，以及如何使用Zookeeper解決實際問題。

Zookeeper基於Java訪問

針對zookeeper，比較常用的Java客戶端有zkclient、curator。由於Curator對於zookeeper的抽象層次比較高，簡化了zookeeper客戶端的開發量。使得curator逐步被廣泛應用。

  1. 封裝zookeeper client與zookeeper server之間的連接處理

  2. 提供了一套fluent風格的操作api

  3. 提供zookeeper各種應用場景（共享鎖、leader選舉）的抽象封裝

依賴jar包

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.2.0</version>
</dependency>

建立連接

curator提供了兩種操作方式來進行操作，一種是Fluent風格，另外一種就是普通的方法調用風格

public class CuratorMain {
    public static void main(String[] args) throws Exception {
        CuratorFramework curatorFramework=
                CuratorFrameworkFactory.newClient("192.168.221.128:2181",5000,20000,
                        new ExponentialBackoffRetry(1000,3));
        curatorFramework.start();
        curatorFramework.blockUntilConnected();
        System.out.println("zookeeper starter success");
        String data=new String(curatorFramework.getData().forPath("/pr"));
        System.out.println("輸出結果："+data);
    }
}

重試策略：Curator內部實現的幾種重試策略:

  • ExponentialBackoffRetry:重試指定的次數, 且每一次重試之間停頓的時間逐漸增加，時間間隔 = baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1)))
  • RetryNTimes:指定最大重試次數的重試策略
  • RetryOneTime:僅重試一次
  • RetryUntilElapsed:一直重試直到達到規定的時間

namespace: 值得注意的是session會話含有隔離命名空間，即客戶端對Zookeeper上數據節點的任何操作都是相對/curator目錄進行的，這有利於實現不同的Zookeeper的業務之間的隔離

public static void main(String[] args) throws Exception {
    CuratorFramework curatorFramework=CuratorFrameworkFactory.builder()
        .connectString("192.168.221.128:2181")
        .sessionTimeoutMs(5000).connectionTimeoutMs(20000)
        .retryPolicy(new ExponentialBackoffRetry(1000,3))
        .namespace("curator").build();
    curatorFramework.start();
    String data=new String(curatorFramework.getData().forPath("/pr"));
    System.out.println("輸出結果："+data);
}

節點的增刪改查

下面代碼演示了Curator訪問Zookeeper實現數據的增刪改查功能。

public class CuratorMain {
    private final CuratorFramework curatorFramework;
    public CuratorMain(){
        curatorFramework=CuratorFrameworkFactory.builder()
            .connectString("192.168.221.128:2181")
            .sessionTimeoutMs(5000).connectionTimeoutMs(20000)
            .retryPolicy(new ExponentialBackoffRetry(1000,3))
            .namespace("curator").build();
        curatorFramework.start();
    }
    public void nodeCRUD() throws Exception {
        System.out.println("開始創建節點");
        String node=curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/node");
        System.out.println("節點創建成功："+node);
        Stat stat=new Stat(); //存儲節點信息
        curatorFramework.getData().storingStatIn(stat).forPath(node);
        System.out.println("查詢節點："+node+"信息，stat:"+stat.toString());
        stat=curatorFramework.setData().withVersion(stat.getVersion()).forPath(node,"Hello World".getBytes());
        String result=new String(curatorFramework.getData().forPath(node));
        System.out.println("修改節點后的數據信息："+result);
        System.out.println("開始刪除節點");
        curatorFramework.delete().forPath(node);
        Stat exist=curatorFramework.checkExists().forPath(node);
        if(exist==null){
            System.out.println("節點刪除成功");
        }
    }

    public static void main(String[] args) throws Exception {
        CuratorMain curatorMain=new CuratorMain();
        curatorMain.nodeCRUD();
    }
}

異步請求

所謂異步請求，就是客戶端發起請求后，由一個異步線程去執行，當收到服務端的返回結果后，再通過回調方法進行通知。

public void asyncCrud() throws Exception {
    CountDownLatch countDownLatch=new CountDownLatch(2);
    ExecutorService executorService= Executors.newFixedThreadPool(2);
    System.out.println("開始節點創建");
    String node=curatorFramework.create().withMode(CreateMode.PERSISTENT).inBackground((session,event)->{
        System.out.println(Thread.currentThread().getName()+"：執行創建節點->"+event.getPath());
        countDownLatch.countDown();
    },executorService).forPath("/async-node");
    System.out.println("異步等待節點創建,此時節點創建狀態，node:"+node);
    curatorFramework.delete().inBackground((session,event)->{
        System.out.println(Thread.currentThread().getName()+"：執行刪除節點->"+event.getPath());
        countDownLatch.countDown();
    },executorService).forPath("/async-node");
    System.out.println("等待異步執行結束");
    countDownLatch.await();
    executorService.shutdown();
}

Zookeeper權限控制

Zookeeper作為一個分布式協調框架，內部存儲了一些分布式系統運行時的狀態的數據，比如master選舉、比如分布式鎖。對這些數據的操作會直接影響到分布式系統的運行狀態。因此，為了保證zookeeper中的數據的安全性，避免誤操作帶來的影響。Zookeeper提供了一套ACL權限控制機制來保證數據的安全。

ACL權限控制，使用：scheme:id:perm來標識。

  • Scheme（權限模式），標識授權策略
  • ID（授權對象）
  • Permission：授予的權限

ZooKeeper的權限控制是基於每個znode節點的，需要對每個節點設置權限，每個znode支持設置多種權限控制方案和多個權限，子節點不會繼承父節點的權限，客戶端無權訪問某節點，但可能可以訪問它的子節點。

Scheme權限模式

Zookeeper提供以下權限模式，所謂權限模式，就是使用什么樣的方式來進行授權。

  • world：默認方式，相當於全部都能訪問。
  • auth：代表已經認證通過的用戶(cli中可以通過addauth digest user:pwd 來添加當前上下文中的授權用戶)
  • digest：即用戶名:密碼這種方式認證，這也是業務系統中最常用的。用 username:password 字符串來產生一個MD5串，然后該串被用來作為ACL ID。認證是通過明文發送username:password 來進行的，當用在ACL時，表達式為username:base64 ，base64是password的SHA1摘要的編碼。
  • ip：通過ip地址來做權限控制，比如 ip:192.168.1.1 表示權限控制都是針對這個ip地址的。也可以針對網段 ip:192.168.1.1/24，此時addr中的有效位與客戶端addr中的有效位進行比對。

ID授權對象

指權限賦予的用戶或一個指定的實體，不同的權限模式下，授權對象不同

img

Id ipId1 = new Id("ip", "192.168.190.1");
Id ANYONE_ID_UNSAFE = new Id("world", "anyone");

Permission權限類型

指通過權限檢查后可以被允許的操作，create /delete /read/write/admin

  • Create 允許對子節點Create 操作

  • Read 允許對本節點GetChildren 和GetData 操作

  • Write 允許對本節點SetData 操作

  • Delete 允許對子節點Delete 操作

  • Admin 允許對本節點setAcl 操作

權限模式(Schema)和授權對象主要用來確認權限驗證過程中使用的驗證策略： 比如ip地址、digest:username:password，匹配到驗證策略並驗證成功后，再根據權限操作類型來決定當前客戶端的訪問權限。

在控制台上實現權限操作

在Zookeeper中提供了ACL相關的命令如下。

getAcl        getAcl <path>     讀取ACL權限
setAcl        setAcl <path> <acl>     設置ACL權限
addauth      addauth <scheme> <auth>     添加認證用戶

word方式

創建一個節點后默認就是world模式

[zk: localhost:2181(CONNECTED) 2] create /auth
Created /auth
[zk: localhost:2181(CONNECTED) 3] getAcl /auth 
'world,'anyone
: cdrwa
[zk: localhost:2181(CONNECTED) 4] create /world
Created /world
[zk: localhost:2181(CONNECTED) 5] getAcl /world
'world,'anyone
: cdrwa
[zk: localhost:2181(CONNECTED) 6] setAcl /world:anyone:acd
setAcl [-s] [-v version] [-R] path acl
[zk: localhost:2181(CONNECTED) 7] setAcl /world world:anyone:acd
[zk: localhost:2181(CONNECTED) 8] getAcl /world
'world,'anyone
: cda

其中， cdrwa，分別對應 create . delete read write admin

IP模式

在ip模式中，首先連接到zkServer的命令需要使用如下方式

./zkCli.sh -server 192.168.221.128:2181

接着按照IP的方式操作如下

[zk: 192.168.221.128:2181(CONNECTED) 3] create /ip-model
Created /ip-model
[zk: 192.168.221.128:2181(CONNECTED) 4] setAcl /ip-model ip:127.0.0.1:cdrwa,ip:192.168.221.128/131:cdrwa
Acl is not valid : /ip-model
[zk: 192.168.221.128:2181(CONNECTED) 5] setAcl /ip-model ip:127.0.0.1:cdrwa,ip:192.168.221.128:cdrwa
[zk: 192.168.221.128:2181(CONNECTED) 6] getAcl /ip-model
'ip,'127.0.0.1
: cdrwa
'ip,'192.168.221.128
: cdrwa

Auth模式

auth模式的操作如下。

[zk: 192.168.221.128:2181(CONNECTED) 7] create /auth
Created /auth
[zk: 192.168.221.128:2181(CONNECTED) 8] addauth digest mic:mic  # 增加授權用戶，明文用戶名和密碼，zk會對密碼加密
[zk: 192.168.221.128:2181(CONNECTED) 9] setAcl /auth auth:mic:cdrwa  # 授予權限
[zk: 192.168.221.128:2181(CONNECTED) 11] getAcl /auth
'digest,'mic:xUsfnPBF9eNvHVWZx/TZt9ioxBA=
: cdrwa
[zk: 192.168.221.128:2181(CONNECTED) 12]

當我們退出當前的會話后，再次連接，執行如下操作，會提示沒有權限

[zk: localhost:2181(CONNECTED) 1] get /auth 
Insufficient permission : /auth

這時候，我們需要重新授權。

[zk: localhost:2181(CONNECTED) 2] addauth digest mic:mic
[zk: localhost:2181(CONNECTED) 3] get /auth
null
[zk: localhost:2181(CONNECTED) 4]

Digest

使用語法，會發現使用方式和Auth模式相同。

setAcl /digest digest:用戶名:密碼:權限

但是有一個不一樣的點，密碼需要用加密后的，否則無法被識別。

密碼： 用戶名和密碼加密后的字符串。

使用下面程序生成密碼

public static void main(String[] args) throws Exception {
    String up="mic:mic";
    byte[] digest=MessageDigest.getInstance("SHA1").digest(up.getBytes());
    String encodeString=Base64.getEncoder().encodeToString(digest);
    System.out.println(encodeString);
}

得到：xUsfnPBF9eNvHVWZx/TZt9ioxBA=

再回到client上進行如下操作

[zk: localhost:2181(CONNECTED) 10] create /digest
Created /digest
[zk: localhost:2181(CONNECTED) 11] setAcl /digest digest:mic:xUsfnPBF9eNvHVWZx/TZt9ioxBA=:cdrwa
[zk: localhost:2181(CONNECTED) 12] getAcl /digest
'digest,'mic:xUsfnPBF9eNvHVWZx/TZt9ioxBA=
: cdrwa

當退出當前會話后，需要再次授權才能訪問/digest節點

[zk: localhost:2181(CONNECTED) 1] get /digest 
Insufficient permission : /digest
[zk: localhost:2181(CONNECTED) 2] addauth digest mic:mic
[zk: localhost:2181(CONNECTED) 3] get /digest
null

Curator演示ACL的使用

接下來我們使用Curator簡單演示一下ACL權限的訪問操作。

public class CuratorMain {
    private final CuratorFramework curatorFramework;
    public CuratorMain(){
        curatorFramework=CuratorFrameworkFactory.builder()
            .connectString("192.168.221.128:2181")
            .sessionTimeoutMs(5000).connectionTimeoutMs(20000)
            .retryPolicy(new ExponentialBackoffRetry(1000,3))
            .namespace("curator").build();
        curatorFramework.start();
    }
    public void aclExample() throws Exception {
        Id id=new Id("digest", DigestAuthenticationProvider.generateDigest("mic:mic"));
        List<ACL> acls=new ArrayList<>();
        acls.add(new ACL(ZooDefs.Perms.ALL,id));
        String node=curatorFramework.create().creatingParentsIfNeeded()
            .withMode(CreateMode.PERSISTENT)
            .withACL(acls,false).forPath("/auth","Hello".getBytes());
        System.out.println("成功創建帶權限的節點:"+node);
        String data=new String(curatorFramework.getData().forPath(node));
        System.out.println("獲取數據結果："+data);
    }
    public static void main(String[] args) throws Exception {
        CuratorMain curatorMain=new CuratorMain();
        curatorMain.aclExample();
    }
}

上述代碼執行后會報錯

Exception in thread "main" org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /curator/auth

修改后代碼如下。

public class CuratorMain {
    private final CuratorFramework curatorFramework;
    public CuratorMain(){

        curatorFramework=CuratorFrameworkFactory.builder()
            .connectString("192.168.221.128:2181")
            .sessionTimeoutMs(20000).connectionTimeoutMs(20000)
            .retryPolicy(new ExponentialBackoffRetry(1000,3))
            .authorization("digest","mic:mic".getBytes()) //在連接時增加授權，即可訪問
            .namespace("curator").build();
        curatorFramework.start();
    }
    public void aclExample() throws Exception {
        Id id=new Id("digest", DigestAuthenticationProvider.generateDigest("mic:mic"));
        List<ACL> acls=new ArrayList<>();
        acls.add(new ACL(ZooDefs.Perms.ALL,id));
        String node=curatorFramework.create().creatingParentsIfNeeded()
            .withMode(CreateMode.PERSISTENT)
            .withACL(acls,false).forPath("/auth","Hello".getBytes());
        System.out.println("成功創建帶權限的節點:"+node);
        String data=new String(curatorFramework.getData().forPath(node));
        System.out.println("獲取數據結果："+data);
    }

    public static void main(String[] args) throws Exception {
        CuratorMain curatorMain=new CuratorMain();
        curatorMain.aclExample();
    }
}

事件監聽機制詳解

在上一節課中，我們了解了Zookeeper中的事件監聽機制，基於事件監聽，應用程序可以訂閱指定節點的變更事件來完成響應的邏輯，這個特性可以讓zookeeper實現分布式鎖、注冊中心、配置中心等功能。

在Zookeeper客戶端中，提供了一下以下事件類型

public static enum EventType {
   //當zookeeper客戶端的連接狀態發生變更時，即KeeperState.Expired、KeeperState.Disconnected、KeeperState.SyncConnected、KeeperState.AuthFailed狀態
    None(-1),         
   //當node-x這個節點被創建時，該事件被觸發
    NodeCreated(1),
   //當node-x這個節點被刪除時，該事件被觸發。 
    NodeDeleted(2),
   //當node-x這個節點的數據發生變更時，該事件被觸發
    NodeDataChanged(3),
   //當node-x這個節點的直接子節點被創建、被刪除、子節點數據發生變更時，該事件被觸發。
    NodeChildrenChanged(4),
   //當node-x這個節點的訂閱事件被移除時
    DataWatchRemoved(5),
   //當node-x這個節點的直接子節點的事件被移除時
    ChildWatchRemoved(6),
   //當值就訂閱被移除時
    PersistentWatchRemoved(7);
}

在zookeeper3.6版本之前，Curator提供了三種Watcher來監聽節點的變化。

  • PathChildCache：監視一個路徑下子結點的創建、刪除、更新。
  • NodeCache：監視當前結點的創建、更新、刪除，並將結點的數據緩存在本地。
  • TreeCache：PathChildCache和NodeCache的“合體”，監視路徑下的創建、更新、刪除事件，並緩存路徑下所有孩子結點的數據。

但是在zookeeper3.6版本之后，只提供了一個CuratorCache來實現時間訂閱。當然，如果要使用事件訂閱功能，我們需要引入下面的jar包。

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.2.0</version>
</dependency>

普通事件訂閱

普通的事件訂閱，就是使用如getData、exists等命令添加的CuratorWatcher機制。這種方式觸發的事件，只會響應一次。

public class CuratorWatchMain {
    private final CuratorFramework curatorFramework;
    public CuratorWatchMain(){

        curatorFramework=CuratorFrameworkFactory.builder()
            .connectString("192.168.221.128:2181")
            .sessionTimeoutMs(20000).connectionTimeoutMs(20000)
            .retryPolicy(new ExponentialBackoffRetry(1000,3))
            .namespace("curator").build();
        curatorFramework.start();
    }

    public void normalWatcher() throws Exception {
        CuratorWatcher watcher=new CuratorWatcher() {
            @Override
            public void process(WatchedEvent watchedEvent) throws Exception {
                System.out.println("監聽事件："+watchedEvent.toString());
            }
        };
        String node=curatorFramework.create().forPath("/listener","I'Listener".getBytes());
        //設置對當前節點的修改和刪除事件
        String data=new String(curatorFramework.getData().usingWatcher(watcher).forPath(node));
        System.out.println(node+"節點對應的value："+data);
        //第一次更新
        curatorFramework.setData().forPath(node,"change listener".getBytes());
        //第二次更新
        curatorFramework.setData().forPath(node,"change listener".getBytes());
    }

    public static void main(String[] args) throws Exception {
        CuratorWatchMain curatorMain=new CuratorWatchMain();
        curatorMain.normalWatcher();
    }
}

如果希望事件監聽是持久化的，則改造代碼如下

public void normalWatcher() throws Exception {
    CuratorWatcher watcher=new CuratorWatcher() {
        @Override
        public void process(WatchedEvent watchedEvent) throws Exception {
            System.out.println("監聽事件："+watchedEvent.toString());
            curatorFramework.checkExists().usingWatcher(this).forPath("/listener");
        }
    };
    String node=curatorFramework.create().forPath("/listener","I'Listener".getBytes());
    //設置對當前節點的修改和刪除事件
    String data=new String(curatorFramework.getData().usingWatcher(watcher).forPath(node));
    System.out.println(node+"節點對應的value："+data);
    //第一次更新
    curatorFramework.setData().forPath(node,"change listener".getBytes());
    Thread.sleep(3000);
    //第二次更新
    curatorFramework.setData().forPath(node,"change listener".getBytes());
    System.in.read();
}

CuratorCache API說明

在Curator包中，提供了另外一個可以持續訂閱的API，CuratorCacheListener

CuratorCacheListener是基於CuratorCache緩存實現的監聽器，CuratorCache對Zookeeper事件監聽進行了封裝，能夠自動處理反復注冊監聽，在使用CuratorListener時，首選需要構建CuratorCache緩存實例，具體定義如下。

CuratorCache.build(CuratorFramework client, String path, Options... options)

Parameters:
client - the client     //客戶端連接
path - path to watch    //需要訂閱事件的節點
options - empty or one or more options

options有三個選項：

CuratorCache.Options.SINGLE_NODE_CACHE

enum Options
{
    /**
    * Normally the entire tree of nodes starting at the given node are cached. This option
    * causes only the given node to be cached (i.e. a single node cache)
    單節點緩存
    */
    SINGLE_NODE_CACHE,

    /**
    * Decompress data via {@link org.apache.curator.framework.api.GetDataBuilder#decompressed()}
    對數據進行壓縮
    */
    COMPRESSED_DATA,
    
    /**
    * Normally, when the cache is closed via {@link CuratorCache#close()}, the storage is cleared
    * via {@link CuratorCacheStorage#clear()}. This option prevents the storage from being cleared.
    關閉后不清理緩存
    */
    DO_NOT_CLEAR_ON_CLOSE
}

CuratorCache實現事件訂閱

代碼實現如下。

public class CuratorWatchMain {
    private final CuratorFramework curatorFramework;
    public CuratorWatchMain(){

        curatorFramework=CuratorFrameworkFactory.builder()
            .connectString("192.168.221.128:2181")
            .sessionTimeoutMs(20000).connectionTimeoutMs(20000)
            .retryPolicy(new ExponentialBackoffRetry(1000,3))
            .authorization("digest","mic:mic".getBytes())
            .namespace("curator").build();
        curatorFramework.start();
    }

    public void normalWatcher() throws Exception {
        CuratorWatcher watcher=new CuratorWatcher() {
            @Override
            public void process(WatchedEvent watchedEvent) throws Exception {
                System.out.println("監聽事件："+watchedEvent.toString());
                curatorFramework.checkExists().usingWatcher(this).forPath("/listener");
            }
        };
        String node=curatorFramework.create().forPath("/listener","I'Listener".getBytes());
        //設置對當前節點的修改和刪除事件
        String data=new String(curatorFramework.getData().usingWatcher(watcher).forPath(node));
        System.out.println(node+"節點對應的value："+data);
        //第一次更新
        curatorFramework.setData().forPath(node,"change listener".getBytes());
        Thread.sleep(3000);
        //第二次更新
        curatorFramework.setData().forPath(node,"change listener".getBytes());
        System.in.read();
    }

    public void addListenerWithNodeCache(String node){
        CuratorCache curatorCache=CuratorCache.build(curatorFramework,node,CuratorCache.Options.SINGLE_NODE_CACHE);
        /**
         * type表示事件類型，NODE_CREATE,NODE_CHANGE,NODE_DELETE
         * forAll: 表示對所有事件的監聽
         * forDeletes: 刪除節點事件
         * forChange:  節點更新事件監聽
         */
        CuratorCacheListener listener=CuratorCacheListener
            .builder()
            .forAll((type, oldNode, newNode)->{
                System.out.println("事件類型："+type+"\n\r原節點："+oldNode+"\n\r新節點"+newNode);
            }).forInitialized(()->{
            System.out.println("初始化");
        }).build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
    }
    public void operation(String node) throws Exception {
        curatorFramework.create().forPath(node);
        curatorFramework.setData().forPath(node,"Hello".getBytes());
        curatorFramework.delete().forPath(node);
    }

    public static void main(String[] args) throws Exception {
        CuratorWatchMain curatorMain=new CuratorWatchMain();
        String node="/node";
        curatorMain.addListenerWithNodeCache(node);
        curatorMain.operation(node);
        System.in.read();
    }
}

關注[跟着Mic學架構]公眾號，獲取更多精品原創


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 springBoot 整合 ZooKeeper Java客戶端之 Apache Curator 實戰 zookeeper（六）：Zookeeper客戶端Curator的API使用詳解 zookeeper生產最廣泛使用java客戶端curator介紹及其它客戶端比較 7.5 zookeeper客戶端curator的基本使用 + zkui zookeeper開源客戶端curator Zookeeper客戶端Curator基本API 基於Apache Curator框架的ZooKeeper使用詳解 Zookeeper系列三：Zookeeper客戶端的使用（Zookeeper原生API如何進行調用、ZKClient、Curator）和Zookeeper會話 基於Apache Curator框架的ZooKeeper使用詳解 zookeeper使用詳解（命令、客戶端、源碼）
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM