Apache Zookeeper Java客户端Curator

作者: 跟着Mic学架构 | 来源:发表于2021-10-31 13:43 被阅读0次

这篇文章是让大家了解Zookeeper基于Java客户端Curator的基本操作，以及如何使用Zookeeper解决实际问题。

Zookeeper基于Java访问

针对zookeeper，比较常用的Java客户端有zkclient、curator。由于Curator对于zookeeper的抽象层次比较高，简化了zookeeper客户端的开发量。使得curator逐步被广泛应用。

封装zookeeper client与zookeeper server之间的连接处理
提供了一套fluent风格的操作api
提供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();
    }
}

Apache Zookeeper Java客户端Curator

Zookeeper基于Java访问

依赖jar包

建立连接

节点的增删改查

异步请求

Zookeeper权限控制

Scheme权限模式

ID授权对象

Permission权限类型

在控制台上实现权限操作

word方式

IP模式

Auth模式

Digest

Curator演示ACL的使用

事件监听机制详解

普通事件订阅

CuratorCache API说明

CuratorCache实现事件订阅

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Apache Zookeeper Java客户端Curator

Zookeeper基于Java访问

依赖jar包

建立连接

节点的增删改查

异步请求

Zookeeper权限控制

Scheme权限模式

ID授权对象

Permission权限类型

在控制台上实现权限操作

word方式

IP模式

Auth模式

Digest

Curator演示ACL的使用

事件监听机制详解

普通事件订阅

CuratorCache API说明

CuratorCache实现事件订阅

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Apache Zookeeper Java客户端Curator