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。

相关文章

Spring Boot + MyBatis Plus 高效开发实战从入门到进阶优化(推荐)

Spring Boot + MyBatis Plus 高效开发实战从入门到进阶优化(推荐)

《SpringBoot+MyBatisPlus高效开发实战从入门到进阶优化(推荐)》本文将详细介绍SpringBoot+MyBatisPlus的完整开发流程,并深入剖析分页查询、批量操作、动... 目录Spring Boot + MyBATis Plus 高效开发实战：从入门到进阶优化1. MyBatis
阅读更多...
SpringCloud动态配置注解@RefreshScope与@Component的深度解析

SpringCloud动态配置注解@RefreshScope与@Component的深度解析

《SpringCloud动态配置注解@RefreshScope与@Component的深度解析》在现代微服务架构中,动态配置管理是一个关键需求,本文将为大家介绍SpringCloud中相关的注解@Re... 目录引言1. @RefreshScope 的作用与原理1.1 什么是 @RefreshScope1.
阅读更多...
Java并发编程必备之Synchronized关键字深入解析

Java并发编程必备之Synchronized关键字深入解析

《Java并发编程必备之Synchronized关键字深入解析》本文我们深入探索了Java中的Synchronized关键字,包括其互斥性和可重入性的特性,文章详细介绍了Synchronized的三种... 目录一、前言二、Synchronized关键字2.1 Synchronized的特性1. 互斥2.
阅读更多...
Spring Boot 配置文件之类型、加载顺序与最佳实践记录

Spring Boot 配置文件之类型、加载顺序与最佳实践记录

《SpringBoot配置文件之类型、加载顺序与最佳实践记录》SpringBoot的配置文件是灵活且强大的工具,通过合理的配置管理,可以让应用开发和部署更加高效,无论是简单的属性配置,还是复杂... 目录Spring Boot 配置文件详解一、Spring Boot 配置文件类型1.1 applicatio
阅读更多...
使用Sentinel自定义返回和实现区分来源方式

使用Sentinel自定义返回和实现区分来源方式

《使用Sentinel自定义返回和实现区分来源方式》：本文主要介绍使用Sentinel自定义返回和实现区分来源方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Sentinel自定义返回和实现区分来源1. 自定义错误返回2. 实现区分来源总结Sentinel自定
阅读更多...
Mysql表的简单操作(基本技能)

Mysql表的简单操作(基本技能)

《Mysql表的简单操作(基本技能)》在数据库中,表的操作主要包括表的创建、查看、修改、删除等,了解如何操作这些表是数据库管理和开发的基本技能,本文给大家介绍Mysql表的简单操作,感兴趣的朋友一起看... 目录3.1 创建表 3.2 查看表结构3.3 修改表3.4 实践案例：修改表在数据库中，表的操作主要
阅读更多...
C# WinForms存储过程操作数据库的实例讲解

C# WinForms存储过程操作数据库的实例讲解

《C#WinForms存储过程操作数据库的实例讲解》：本文主要介绍C#WinForms存储过程操作数据库的实例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、存储过程基础二、C# 调用流程1. 数据库连接配置2. 执行存储过程（增删改）3. 查询数据三、事务处
阅读更多...
Pandas使用SQLite3实战

Pandas使用SQLite3实战

《Pandas使用SQLite3实战》本文主要介绍了Pandas使用SQLite3实战,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学... 目录1 环境准备2 从 SQLite3VlfrWQzgt 读取数据到 DataFrame基础用法：读
阅读更多...
JSON Web Token在登陆中的使用过程

JSON Web Token在登陆中的使用过程

《JSONWebToken在登陆中的使用过程》：本文主要介绍JSONWebToken在登陆中的使用过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录JWT 介绍微服务架构中的 JWT 使用结合微服务网关的 JWT 验证1. 用户登录，生成 JWT2. 自定义过滤
阅读更多...
Java中StopWatch的使用示例详解

Java中StopWatch的使用示例详解

《Java中StopWatch的使用示例详解》stopWatch是org.springframework.util包下的一个工具类,使用它可直观的输出代码执行耗时,以及执行时间百分比,这篇文章主要介绍... 目录stopWatch 是org.springframework.util 包下的一个工具类，使用它
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2