Apache Zookeeper Java客户端Curator使用及权限模式详解
这篇文章是让大家了解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授权对象
指权限赋予的用户或一个指定的实体，不同的权限模式下，授权对象不同
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学架构
链接：https://juejin.cn/post/7022813384303181861