Spring Boot 使用 Curator 操作 ZooKeeper

2024-01-11 08:32
文章标签 java 使用 操作 spring boot zookeeper curator

本文主要是介绍Spring Boot 使用 Curator 操作 ZooKeeper，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

本文主要介绍使用 Curator 访问 ZooKeeper 的一些基本方法，而不仅仅限于指定的 Recipes，你可以使用 Curator API 任意的访问 ZooKeeper。Curator 框架提供了一套高级的 API，简化了 ZooKeeper 的操作。它增加了很多使用 ZooKeeper 开发的特性，可以处理 ZooKeeper 集群复杂的连接管理和重试机制。

三种 zk 客户端对比：

原生apizkclientapache curator
优点1、session会话超时重连；
2、解决watcher反复注册；
3、简化api开发；		1、解决watch注册一次就会失效的问题；
2、api更加简单易用；
3、提供了更多解决方案并且实现简单，比如分布式锁；
4、提供了常用的zk工具类；
缺点1、watch注册一次后会失效；
2、session超时之后没有实现重连机制；
3、异常处理繁琐；
4、只提供了简单的byte[]数组的接口，没有提供针对对象级别的序列化；
5、创建节点时如果节点存在抛出异常，需要自行检查节点是否存在；
6、删除节点无法实现级联删除；		1、异常处理简化，抛出RuntimeException；
2、重试机制比较难用；
3、没有提供各种使用场景的实现；

使用 Curator 需要添加 Maven 依赖（两个模块按需添加，ZooKeeper 版本需要与服务器实例中 ZooKeeper 版本一致）：

<!-- zookeeper -->
<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.4.14</version><exclusions><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion></exclusions>
</dependency>
<!-- curator-framework -->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.2.0</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion></exclusions>
</dependency>
<!-- curator-recipes -->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.2.0</version><exclusions><exclusion><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId></exclusion></exclusions>
</dependency>

可以看出，Curator 功能分两大类，一类是对 ZooKeeper 的一些基本命令的封装，比如增删改查，即 Framework 模块；另一类是他的高级特性，即 Recipes 模块。下面将分两部分分别进行阐述。

1.Curator Framework的使用

Curator 框架通过 CuratorFrameworkFactory 以工厂模式和 builder 模式创建 CuratorFramework 实例。CuratorFramework 实例都是线程安全的，你应该在你的应用中共享同一个 CuratorFramework 实例。工厂方法 newClient() 提供了一个简单方式创建实例。而 Builder 提供了更多的参数控制。一旦你创建了一个 CuratorFramework 实例，你必须调用它的 start() 启动，在应用退出时调用 close() 方法关闭。

下面的例子演示了两种创建Curator的方法：

创建方式一（以 builder 模式创建，Fluent 风格，建议使用这种）：

// 多个地址逗号隔开
CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181").sessionTimeoutMs(1000)    // 连接超时时间.connectionTimeoutMs(1000) // 会话超时时间// 刚开始重试间隔为1秒，之后重试间隔逐渐增加，最多重试不超过三次.retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
client.start();
//...
CloseableUtils.closeQuietly(client);//建议放在finally块中

创建方式二（以工厂模式创建）：

// 多个地址逗号隔开
CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181",// 刚开始重试间隔为1秒，之后重试间隔逐渐增加，最多重试不超过三次new ExponentialBackoffRetry(1000, 3));
client.start();
//...
CloseableUtils.closeQuietly(client); // 建议放在finally块中

下面看下 CuratorFramework 提供的方法：

方法名描述
create()开始创建操作，可以调用额外的方法(比如方式mode或者后台执行background)并在最后调用forPath()指定要操作的ZNode
delete()开始删除操作，可以调用额外的方法(版本或者后台处理version or background)并在最后调用forPath()指定要操作的ZNode
checkExists()开始检查ZNode是否存在的操作，可以调用额外的方法(监控或者后台处理)并在最后调用forPath()指定要操作的ZNode
getData()开始获得ZNode节点数据的操作，可以调用额外的方法(监控、后台处理或者获取状态watch, background or get stat)并在最后调用forPath()指定要操作的ZNode
setData()开始设置ZNode节点数据的操作，可以调用额外的方法(版本或者后台处理)并在最后调用forPath()指定要操作的ZNode
getChildren()开始获得ZNode的子节点列表，以调用额外的方法(监控、后台处理或者获取状态watch, background or get stat)并在最后调用forPath()指定要操作的ZNode
inTransaction()开始是原子ZooKeeper事务，可以复合create, setData, check, and/or delete 等操作然后调用commit()作为一个原子操作提交

后台操作的通知和监控可以通过 ClientListener 接口发布，你可以在 CuratorFramework 实例上通过 addListener() 注册 listener，Listener 实现了下面的方法:

  • eventReceived() 一个后台操作完成或者一个监控被触发

事件类型以及事件的方法如下：

Event TypeEvent Methods
CREATEgetResultCode()、getPath()
DELETEgetResultCode()、getPath()
EXISTSgetResultCode()、getPath()、getStat()
GETDATAgetResultCode()、getPath()、getStat()、getData()
SETDATAgetResultCode()、getPath()、getStat()
CHILDRENgetResultCode()、getPath()、getStat()、getChildren()
WATCHEDgetWatchedEvent()

还可以通过 ConnectionStateListener 接口监控连接的状态，强烈推荐你增加这个监控器。你可以使用命名空间 Namespace 避免多个应用的节点的名称冲突。CuratorFramework 提供了命名空间的概念，这样 CuratorFramework 会为它的 API 调用的 path 加上命名空间：

CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181").namespace("configCenter")   // 命名空间....build();
client.start();
client.create().forPath("/search/business/test", data);
//node was actually written to: "/configCenter/search/business/test"
...
CloseableUtils.closeQuietly(client); // 建议放在finally块中

Curator 还提供了临时的 CuratorFramework: CuratorTempFramework，3分钟不活动连接就被关闭，你也可以指定不活动的时间。创建 builder 时不是调用 build() 而是调用 buildTemp()。CuratorTempFramework 只提供了 close()、inTransaction()、getData() 这三个方法。

下面我们看下 Curator 增删改查 ZooKeeper 的具体写法：

client.start();// 判断是否存在，Stat就是对znode所有属性的一个映射，stat=null表示节点不存在
Stat stat = client.checkExists().forPath("/search/business/test");// 查询子节点
List<String> childNodes = client.getChildren().forPath("/search/business/test"));// 创建节点
client.create().creatingParentsIfNeeded() // 若创建节点的父节点不存在则先创建父节点再创建子节点.withMode(CreateMode.PERSISTENT) // 创建的是持久节点.withACL(Ids.OPEN_ACL_UNSAFE) // 默认匿名权限,权限scheme id:'world,'anyone,:cdrwa.forPath("/search/business/test","1".getBytes());
// 更新节点数据
client.setData().withVersion(2) // 乐观锁.forPath("/search/business/test","0".getBytes());// 读取节点数据
Stat stat = new Stat(); // Stat就是对znode所有属性的一个映射，stat=null表示节点不存在
String re = new String(client.getData().storingStatIn(stat) // 在获取节点内容的同时把状态信息存入Stat对象，如果不写的话只会读取节点数据.forPath("/search/business/test"));// 删除节点
client.delete().guaranteed() // 保障机制，若未删除成功，只要会话有效会在后台一直尝试删除.deletingChildrenIfNeeded() // 若当前节点包含子节点，子节点也删除.withVersion(2) // 指定版本号.forPath("/search/business/test");// watcher事件,使用usingWatcher的时候,监听只会触发一次，监听完毕后就销毁
client.getData().usingWatcher(new CuratorWatcher() {@Overridepublic void process(WatchedEvent event) throws Exception {logger.info("触发watcher, path:{}", event.getPath());}}).forPath("/search/business/test");// watcher事件, NodeCache一次注册N次监, 缺点:没法监听当前节点增删改操作,所以引出了PathChildrenCache
final NodeCache nodeCache = new NodeCache(client, "/search/business/test");
nodeCache.start(true); // 当zkServer与客户端链接的时候, NodeCache会把zk数据缓存到我们本地
if (nodeCache.getCurrentData() != null) {logger.info("节点初始化数据为:{}", new String(nodeCache.getCurrentData().getData()));
} else {logger.info("节点初始化数据为空");
}
nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {if (nodeCache.getCurrentData() != null) {String re = new String(nodeCache.getCurrentData().getData());logger.info("节点路径:{}, 节点数据:{}", nodeCache.getCurrentData().getPath(), re);} else {logger.info("当前节点被删除了");}}
});// 监听父节点以下所有的子节点, 当子节点发生变化的时候(增删改)都会监听到
// 为子节点添加watcher事件
// PathChildrenCache监听数据节点的增删改
final PathChildrenCache childrenCache = new PathChildrenCache(client, "/search", true);
// NORMAL:异步初始化, BUILD_INITIAL_CACHE:同步初始化, POST_INITIALIZED_EVENT:异步初始化,初始化之后会触发事件
childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
List<ChildData> childDataList = childrenCache.getCurrentData(); // 当前数据节点的子节点数据列表
childrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {if (event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)) {logger.info("子节点初始化ok..");} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {logger.info("添加子节点, path:{}, data:{}", event.getData().getPath(), event.getData().getData());} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {logger.info("删除子节点, path:{}", event.getData().getPath());} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {logger.info("修改子节点, path:{}, data:{}", event.getData().getPath(), event.getData().getData());}}
});CloseableUtils.closeQuietly(client);//建议放在finally块中

所有这些方法都以 forpath() 结尾，辅以 watch (监听)，withMode (指定模式)，和 inBackground (后台运行) 等方法来使用。

此外，Curator 还支持事务，一组crud操作要么都完成，要么都不完成：

client.start();// 开启事务
CuratorTransaction transaction = client.inTransaction();
Collection<CuratorTransactionResult> results = transaction.create().forPath("/curator/1", "0".getBytes()).and().setData().forPath("/curator/2", "-1".getBytes()).and().delete().forPath("/curator/3").and().commit();CloseableUtils.closeQuietly(client);//建议放在finally块中

2.Curator Recipes的使用

Recipes 模块主要有 Elections (选举)、Locks (锁)、Barriers (关卡)、Atomic (原子量)、Caches、Queues 等。

1.Elections

选举主要依赖于 LeaderSelector 和 LeaderLatch 两个类。前者是所有存活的客户端不间断的轮流做 Leader。后者是一旦选举出 Leader，除非有客户端挂掉重新触发选举，否则不会交出领导权。这两者在实现上是可以切换的。

下面看一下 LeaderSelector 的选举:

/*** 本类基于leaderSelector实现,所有存活的client会公平的轮流做leader* 如果不想频繁的变化Leader，需要在takeLeadership方法里阻塞leader的变更！ 或者使用 {@link}* LeaderLatchClient*/
public class LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {private static final Log log = LogFactory.getLog(LeaderSelectorClient.class);private final String name;private final LeaderSelector leaderSelector;private final String PATH = "/leaderselector";public LeaderSelectorClient(CuratorFramework client, String name) {this.name = name;leaderSelector = new LeaderSelector(client, PATH, this);leaderSelector.autoRequeue();}public void start() throws IOException {leaderSelector.start();}@Overridepublic void close() throws IOException {leaderSelector.close();}/*** client成为leader后，会调用此方法*/@Overridepublic void takeLeadership(CuratorFramework client) throws Exception {int waitSeconds = (int) (5 * Math.random()) + 1;log.info(name + "是当前的leader");try {Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));} catch (InterruptedException e) {Thread.currentThread().interrupt();} finally {log.info(name + "让出领导权\n");}}
}

2.Locks

Curator Lock 相关的实现在 recipes.locks 包里。顶级接口都是 InterProcessLock。我们直接看最有代表性的 InterProcessReadWriteLock 进程内部读写锁（可重入读写锁）。什么叫可重入，什么叫读写锁。不清楚的先查好资料吧。总之读写锁一定是成对出现的。 我们先定义两个任务，可并行的执行的，和互斥执行的。

参考：
http://www.cnblogs.com/hzhuxin/archive/2012/11/01/2749341.html
http://macrochen.iteye.com/blog/1366136

这篇关于Spring Boot 使用 Curator 操作 ZooKeeper的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/593751。

相关文章

Java如何根据word模板导出数据

Java如何根据word模板导出数据

《Java如何根据word模板导出数据》这篇文章主要为大家详细介绍了Java如何实现根据word模板导出数据,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... pom.XML文件导入依赖 <dependency> <groupId>cn.afterturn</groupId>
阅读更多...
Java应用如何防止恶意文件上传

Java应用如何防止恶意文件上传

《Java应用如何防止恶意文件上传》恶意文件上传可能导致服务器被入侵,数据泄露甚至服务瘫痪,因此我们必须采取全面且有效的防范措施来保护Java应用的安全,下面我们就来看看具体的实现方法吧... 目录恶意文件上传的潜在风险常见的恶意文件上传手段防范恶意文件上传的关键策略严格验证文件类型检查文件内容控制文件存储
阅读更多...
Go语言使用slices包轻松实现排序功能

Go语言使用slices包轻松实现排序功能

《Go语言使用slices包轻松实现排序功能》在Go语言开发中,对数据进行排序是常见的需求,Go1.18版本引入的slices包提供了简洁高效的排序解决方案,支持内置类型和用户自定义类型的排序操作,本... 目录一、内置类型排序：字符串与整数的应用1. 字符串切片排序2. 整数切片排序二、检查切片排序状态：
阅读更多...
浅析Java如何保护敏感数据

浅析Java如何保护敏感数据

《浅析Java如何保护敏感数据》在当今数字化时代,数据安全成为了软件开发中至关重要的课题,本文将深入探讨Java安全领域,聚焦于敏感数据保护的策略与实践,感兴趣的小伙伴可以了解下... 目录一、Java 安全的重要性二、敏感数据加密技术（一）对称加密（二）非对称加密三、敏感数据的访问控制（一）基于角色的访问
阅读更多...
Java计算经纬度距离的示例代码

Java计算经纬度距离的示例代码

《Java计算经纬度距离的示例代码》在Java中计算两个经纬度之间的距离,可以使用多种方法（代码示例均返回米为单位）,文中整理了常用的5种方法,感兴趣的小伙伴可以了解一下... 目录1. Haversine公式（中等精度，推荐通用场景）2. 球面余弦定理（简单但精度较低）3. Vincenty公式（高精度，
阅读更多...
使用Java将实体类转换为JSON并输出到控制台的完整过程

使用Java将实体类转换为JSON并输出到控制台的完整过程

《使用Java将实体类转换为JSON并输出到控制台的完整过程》在软件开发的过程中,Java是一种广泛使用的编程语言,而在众多应用中,数据的传输和存储经常需要使用JSON格式,用Java将实体类转换为J... 在软件开发的过程中，Java是一种广泛使用的编程语言，而在众多应用中，数据的传输和存储经常需要使用j
阅读更多...
Java实现视频格式转换的完整指南

Java实现视频格式转换的完整指南

《Java实现视频格式转换的完整指南》在Java中实现视频格式的转换,通常需要借助第三方工具或库,因为视频的编解码操作复杂且性能需求较高,以下是实现视频格式转换的常用方法和步骤,需要的朋友可以参考下... 目录核心思路方法一：通过调用 FFmpeg 命令步骤示例代码说明优点方法二：使用 Jaffree（FF
阅读更多...
Java实现图片淡入淡出效果

Java实现图片淡入淡出效果

《Java实现图片淡入淡出效果》在现代图形用户界面和游戏开发中,**图片淡入淡出（FadeIn／Out）**是一种常见且实用的视觉过渡效果,它可以用于启动画面、场景切换、轮播图、提示框弹出等场景,通过... 目录1. 项目背景详细介绍2. 项目需求详细介绍2.1 功能需求2.2 非功能需求3. 相关技术详细
阅读更多...
Java如何用乘号来重复字符串的功能

Java如何用乘号来重复字符串的功能

《Java如何用乘号来重复字符串的功能》：本文主要介绍Java使用乘号来重复字符串的功能,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Java乘号来重复字符串的功能1、利用循环2、使用StringBuilder3、采用 Java 11 引入的String.rep
阅读更多...
Mysql数据库中数据的操作CRUD详解

Mysql数据库中数据的操作CRUD详解

《Mysql数据库中数据的操作CRUD详解》：本文主要介绍Mysql数据库中数据的操作（CRUD）,详细描述对Mysql数据库中数据的操作（CRUD）,包括插入、修改、删除数据,还有查询数据,包括... 目录一、插入数据（insert）1.插入数据的语法2.注意事项二、修改数据（update）1.语法2.有
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2