VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > 编程开发 > Java教程 >
  • Apache Zookeeper Java客户端Curator使用及权限模式详解

这篇文章是让大家了解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();
    }
}

来源:https://www.cnblogs.com/mic112/p/15457072.html


相关教程