网站建立与推广,wordpress动态cdn,专业的seo搜索引擎优化培训,えっちな秘密基地视频简介
Curator是Netflix公司开源的一套zookeeper客户端框架#xff0c;解决了很多Zookeeper客户端非常底层的细节开发工作#xff0c;包括连接重连、反复注册Watcher和NodeExistsException异常等等。Patrixck Hunt#xff08;Zookeeper#xff09;以一句“Guava is to Java…简介
Curator是Netflix公司开源的一套zookeeper客户端框架解决了很多Zookeeper客户端非常底层的细节开发工作包括连接重连、反复注册Watcher和NodeExistsException异常等等。Patrixck HuntZookeeper以一句“Guava is to Java that Curator to Zookeeper”给Curator予高度评价。 Curator的maven依赖
Apache Curator 是 Apache 基金会提供的一款 ZooKeeper 客户端它提供了一套易用性和可读性非常强的 Fluent 风格的客户端 API 可以帮助我们快速搭建稳定可靠的 ZooKeeper 客户端程序。
为便于你更全面了解 Curator 的功能我整理出了如下表格展示了 Curator 提供的 jar 包 添加依赖
!-- 对zookeeper的底层api的一些封装 --dependencygroupIdorg.apache.curator/groupIdartifactIdcurator-framework/artifactIdversion5.0.0/version/dependency!-- 封装了一些高级特性如Cache事件监听、选举、分布式锁、分布式Barrier --dependencygroupIdorg.apache.curator/groupIdartifactIdcurator-recipes/artifactIdversion5.0.0/version/dependency创建连接
public static void main(String[] args) throws Exception {//fluent风格
// CuratorFramework curatorFramework1CuratorFrameworkFactory.builder().connectString(ZKSERVERS).sessionTimeoutMs(5000).
// retryPolicy(new ExponentialBackoffRetry(1000,3)).
// namespace(/curator).build();
// curatorFramework1.start();RetryPolicy retryPolicy new ExponentialBackoffRetry(1000, 3);CuratorFramework client CuratorFrameworkFactory.newClient(ZKSERVERS, retryPolicy);client.start();String resultclient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(/curator/curator1/curator11,123.getBytes());System.out.println(result);}curator连接的重试策略
ExponentialBackoffRetry() 衰减重试 RetryNTimes 指定最大重试次数 RetryOneTime 仅重试一次 RetryUnitilElapsed 一直重试直到规定的时间 基本操作
public static void main(String[] args) throws Exception {// Zookeeper集群地址多个节点地址可以用逗号分隔String zkAddress 127.0.0.1:2181;// 重试策略如果连接不上ZooKeeper集群会重试三次重试间隔会递增RetryPolicy retryPolicy new ExponentialBackoffRetry(1000, 3);// 创建Curator Client并启动启动成功之后就可以与Zookeeper进行交互了CuratorFramework client CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);client.start();// 下面简单说明Curator中常用的API// create()方法创建ZNode可以调用额外方法来设置节点类型、添加Watcher// 下面是创建一个名为user的持久节点其中会存储一个test字符串String path client.create().withMode(CreateMode.PERSISTENT).forPath(/user, test.getBytes());System.out.println(path);// 输出:/user// checkExists()方法可以检查一个节点是否存在Stat stat client.checkExists().forPath(/user);System.out.println(stat!null);// 输出:true返回的Stat不为null即表示节点存在// getData()方法可以获取一个节点中的数据byte[] data client.getData().forPath(/user);System.out.println(new String(data));// 输出:test// setData()方法可以设置一个节点中的数据stat client.setData().forPath(/user,data.getBytes());data client.getData().forPath(/user);System.out.println(new String(data));// 输出:data// 在/user节点下创建多个临时顺序节点for (int i 0; i 3; i) {client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(/user/child-);}// 获取所有子节点ListString children client.getChildren().forPath(/user);System.out.println(children);// 输出[child-0000000002, child-0000000001, child-0000000000]// delete()方法可以删除指定节点deletingChildrenIfNeeded()方法// 会级联删除子节点client.delete().deletingChildrenIfNeeded().forPath(/user);}Curator 异步接口引入了BackgroundCallback
上面介绍的创建、删除、更新、读取等方法都是同步的Curator 提供异步接口引入了BackgroundCallback 这个回调接口以及 CuratorListener 这个监听器用于处理 Background 调用之后服务端返回的结果信息。BackgroundCallback 接口和 CuratorListener 监听器中接收一个 CuratorEvent 的参数里面包含事件类型、响应码、节点路径等详细信息。
下面我们通过一个示例说明 BackgroundCallback 接口以及 CuratorListener 监听器的基本使用
public class CuratorAsynApi {public static void main(String[] args) throws Exception {// Zookeeper集群地址多个节点地址可以用逗号分隔String zkAddress 127.0.0.1:2181;// 重试策略如果连接不上ZooKeeper集群会重试三次重试间隔会递增RetryPolicy retryPolicy new ExponentialBackoffRetry(1000,3);// 创建Curator Client并启动启动成功之后就可以与Zookeeper进行交互了CuratorFramework client CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);client.start();// 添加CuratorListener监听器针对不同的事件进行处理client.getCuratorListenable().addListener(new CuratorListener() {public void eventReceived(CuratorFramework client,CuratorEvent event) throws Exception {switch (event.getType()) {case CREATE:System.out.println(CREATE: event.getPath());break;case DELETE:System.out.println(DELETE: event.getPath());break;case EXISTS:System.out.println(EXISTS: event.getPath());break;case GET_DATA:System.out.println(GET_DATA: event.getPath() , new String(event.getData()));break;case SET_DATA:System.out.println(SET_DATA: new String(event.getData()));break;case CHILDREN:System.out.println(CHILDREN: event.getPath());break;default:}}});// 注意:下面所有的操作都添加了inBackground()方法转换为后台操作client.create().withMode(CreateMode.PERSISTENT).inBackground().forPath(/user, test.getBytes());client.checkExists().inBackground().forPath(/user);client.setData().inBackground().forPath(/user,setData-Test.getBytes());client.getData().inBackground().forPath(/user);for (int i 0; i 3; i) {client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).inBackground().forPath(/user/child-);}client.getChildren().inBackground().forPath(/user);// 添加BackgroundCallbackclient.getChildren().inBackground(new BackgroundCallback() {public void processResult(CuratorFramework client,CuratorEvent event) throws Exception {System.out.println(in background: event.getType() , event.getPath());}}).forPath(/user);client.delete().deletingChildrenIfNeeded().inBackground().forPath(/user);System.in.read();}}连接状态监听
除了基础的数据操作Curator 还提供了监听连接状态的监听器——ConnectionStateListener它主要是处理 Curator 客户端和 ZooKeeper 服务器间连接的异常情况例如 短暂或者长时间断开连接。
短暂断开连接时ZooKeeper 客户端会检测到与服务端的连接已经断开但是服务端维护的客户端 Session 尚未过期之后客户端和服务端重新建立了连接当客户端重新连接后由于 Session 没有过期ZooKeeper 能够保证连接恢复后保持正常服务。
而长时间断开连接时Session 已过期与先前 Session 相关的 Watcher 和临时节点都会丢失。当 Curator 重新创建了与 ZooKeeper 的连接时会获取到 Session 过期的相关异常Curator 会销毁老 Session并且创建一个新的 Session。由于老 Session 关联的数据不存在了在 ConnectionStateListener 监听到 LOST 事件时就可以依靠本地存储的数据恢复 Session 了。
这里 Session 指的是 ZooKeeper 服务器与客户端的会话。客户端启动的时候会与服务器建立一个 TCP 连接从第一次连接建立开始客户端会话的生命周期也开始了。客户端能够通过心跳检测与服务器保持有效的会话也能够向 ZooKeeper 服务器发送请求并接受响应同时还能够通过该连接接收来自服务器的 Watch 事件通知。
我们可以设置客户端会话的超时时间sessionTimeout当服务器压力太大、网络故障或是客户端主动断开连接等原因导致连接断开时只要客户端在 sessionTimeout 规定的时间内能够重新连接到 ZooKeeper 集群中任意一个实例那么之前创建的会话仍然有效。ZooKeeper 通过 sessionID 唯一标识 Session所以在 ZooKeeper 集群中sessionID 需要保证全局唯一。 由于 ZooKeeper 会将 Session 信息存放到硬盘中即使节点重启之前未过期的 Session 仍然会存在。
public class CuratorSessionApi {public static void main(String[] args) throws Exception {// Zookeeper集群地址多个节点地址可以用逗号分隔String zkAddress 127.0.0.1:2181;// 重试策略如果连接不上ZooKeeper集群会重试三次重试间隔会递增RetryPolicy retryPolicy new ExponentialBackoffRetry(1000,3);// 创建Curator Client并启动启动成功之后就可以与Zookeeper进行交互了CuratorFramework client CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);client.start();// 添加ConnectionStateListener监听器client.getConnectionStateListenable().addListener(new ConnectionStateListener() {public void stateChanged(CuratorFramework client,ConnectionState newState) {// 这里我们可以针对不同的连接状态进行特殊的处理switch (newState) {case CONNECTED:// 第一次成功连接到ZooKeeper之后会进入该状态。// 对于每个CuratorFramework对象此状态仅出现一次System.out.println(第一次成功连接到ZooKeeper之后会进入该状态);break;case SUSPENDED: // ZooKeeper的连接丢失System.out.println(ZooKeeper的连接丢失);break;case RECONNECTED: // 丢失的连接被重新建立System.out.println(丢失的连接被重新建立);break;case LOST:System.out.println(当Curator认为会话已经过期时则进入此状态);// 当Curator认为会话已经过期时则进入此状态break;case READ_ONLY: // 连接进入只读模式System.out.println(连接进入只读模式);break;}}});client.close();System.in.read();}
}Watcher
Watcher 监听机制是 ZooKeeper 中非常重要的特性可以监听某个节点上发生的特定事件例如监听节点数据变更、节点删除、子节点状态变更等事件。当相应事件发生时ZooKeeper 会产生一个 Watcher 事件并且发送到客户端。通过 Watcher 机制就可以使用 ZooKeeper 实现分布式锁、集群管理等功能。
在 Curator 客户端中我们可以使用 usingWatcher() 方法添加 Watcher前面示例中能够添加 Watcher 的有 checkExists()、getData()以及 getChildren() 三个方法下面我们来看一个具体的示例
public class CuratorWatcherApi {public static void main(String[] args) throws Exception {// Zookeeper集群地址多个节点地址可以用逗号分隔String zkAddress 127.0.0.1:2181;// 重试策略如果连接不上ZooKeeper集群会重试三次重试间隔会递增RetryPolicy retryPolicy new ExponentialBackoffRetry(1000,3);// 创建Curator Client并启动启动成功之后就可以与Zookeeper进行交互了CuratorFramework client CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);client.start();try {client.create().withMode(CreateMode.PERSISTENT).forPath(/user, test.getBytes());} catch (Exception e) {}// 这里通过usingWatcher()方法添加一个WatcherListString children client.getChildren().usingWatcher(new CuratorWatcher() {public void process(WatchedEvent event) throws Exception {System.out.println(event.getType() , event.getPath());}}).forPath(/user);System.out.println(children);System.in.read();}
}接下来我们打开 ZooKeeper 的命令行客户端在 /user 节点下先后添加两个子节点此时我们只得到一行输出 NodeChildrenChanged,/user 之所以这样是因为通过 usingWatcher() 方法添加的 CuratorWatcher 只会触发一次触发完毕后就会销毁。checkExists() 方法、getData() 方法通过 usingWatcher() 方法添加的 Watcher 也是一样的原理只不过监听的事件不同你若感兴趣的话可以自行尝试一下。
相信你已经感受到直接通过注册 Watcher 进行事件监听不是特别方便需要我们自己反复注册 Watcher。Apache Curator 引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。Cache 是 Curator 中对事件监听的包装其对事件的监听其实可以近似看作是一个本地缓存视图和远程ZooKeeper 视图的对比过程。同时Curator 能够自动为开发人员处理反复注册监听从而大大简化了代码的复杂程度。
实践中常用的 Cache 有三大类
NodeCache。 对一个节点进行监听监听事件包括指定节点的增删改操作。注意哦NodeCache 不仅可以监听数据节点的内容变更也能监听指定节点是否存在如果原本节点不存在那么 Cache 就会在节点被创建后触发 NodeCacheListener删除操作亦然。PathChildrenCache。 对指定节点的一级子节点进行监听监听事件包括子节点的增删改操作但是不对该节点的操作监听。TreeCache。 综合 NodeCache 和 PathChildrenCache 的功能是对指定节点以及其子节点进行监听同时还可以设置监听的深度。 下面通过示例介绍上述三种 Cache 的基本使用
public class CuratorWatcherCacheApi {public static void main(String[] args) throws Exception {// Zookeeper集群地址多个节点地址可以用逗号分隔String zkAddress 127.0.0.1:2181;// 重试策略如果连接不上ZooKeeper集群会重试三次重试间隔会递增RetryPolicy retryPolicy new ExponentialBackoffRetry(1000,3);// 创建Curator Client并启动启动成功之后就可以与Zookeeper进行交互了CuratorFramework client CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);client.start();// 创建NodeCache监听的是/user这个节点NodeCache nodeCache new NodeCache(client, /user);// start()方法有个boolean类型的参数默认是false。如果设置为true// 那么NodeCache在第一次启动的时候就会立刻从ZooKeeper上读取对应节点的// 数据内容并保存在Cache中。nodeCache.start(true);if (nodeCache.getCurrentData() ! null) {System.out.println(NodeCache节点初始化数据为 new String(nodeCache.getCurrentData().getData()));} else {System.out.println(NodeCache节点数据为空);}// 添加监听器nodeCache.getListenable().addListener(() - {String data new String(nodeCache.getCurrentData().getData());System.out.println(NodeCache节点路径 nodeCache.getCurrentData().getPath() 节点数据为 data);});// 创建PathChildrenCache实例监听的是user这个节点PathChildrenCache childrenCache new PathChildrenCache(client, /user, true);// StartMode指定的初始化的模式// NORMAL:普通异步初始化// BUILD_INITIAL_CACHE:同步初始化// POST_INITIALIZED_EVENT:异步初始化初始化之后会触发事件childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);// childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);// childrenCache.start(PathChildrenCache.StartMode.NORMAL);ListChildData children childrenCache.getCurrentData();System.out.println(获取子节点列表);// 如果是BUILD_INITIAL_CACHE可以获取这个数据如果不是就不行children.forEach(childData - {System.out.println(new String(childData.getData()));});childrenCache.getListenable().addListener(((client1, event) - {System.out.println(LocalDateTime.now() event.getType());if (event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)) {System.out.println(PathChildrenCache:子节点初始化成功...);} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {String path event.getData().getPath();System.out.println(PathChildrenCache添加子节点: event.getData().getPath());System.out.println(PathChildrenCache子节点数据: new String(event.getData().getData()));} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {System.out.println(PathChildrenCache删除子节点: event.getData().getPath());} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {System.out.println(PathChildrenCache修改子节点路径: event.getData().getPath());System.out.println(PathChildrenCache修改子节点数据: new String(event.getData().getData()));}}));// 创建TreeCache实例监听user节点TreeCache cache TreeCache.newBuilder(client, /user).setCacheData(false).build();cache.getListenable().addListener((c, event) - {if (event.getData() ! null) {System.out.println(TreeCache,type event.getType() path event.getData().getPath());} else {System.out.println(TreeCache,type event.getType());}});cache.start();System.in.read();}
}启动程序后自己在客户端进行增加节点修改数据等操作观察输出这里不进行截图了
