-
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();
}
}
来源：https://www.cnblogs.com/mic112/p/15457072.html
最新更新
python爬虫及其可视化
使用python爬取豆瓣电影短评评论内容
nodejs爬虫
Python正则表达式完全指南
爬取豆瓣Top250图书数据
shp 地图文件批量添加字段
爬虫小试牛刀（爬取学校通知公告）
【python基础】函数-初识函数
【python基础】函数-返回值
HTTP请求：requests模块基础使用必知必会
SQL SERVER中递归
2个场景实例讲解GaussDB(DWS)基表统计信息估
常用的 SQL Server 关键字及其含义
动手分析SQL Server中的事务中使用的锁
openGauss内核分析：SQL by pass & 经典执行
一招教你如何高效批量导入与更新数据
天天写SQL，这些神奇的特性你知道吗？
openGauss内核分析：执行计划生成
[IM002]Navicat ODBC驱动器管理器 未发现数据
初入Sql Server 之 存储过程的简单使用
uniapp/H5 获取手机桌面壁纸 (静态壁纸）
[前端] DNS解析与优化
为什么在js中需要添加addEventListener（）？
JS模块化系统
js通过Object.defineProperty() 定义和控制对象
这是目前我见过最好的跨域解决方案！
减少回流与重绘
减少回流与重绘
如何使用KrpanoToolJS在浏览器切图
performance.now() 与 Date.now() 对比