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。

相关文章

JVM 的类初始化机制

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时，有没有考虑过 JVM 是如何把静态的字节码（byte code）转化为运行时对象的呢，这个问题看似简单，但清楚的同学相信也不会太多，这篇文章首先介绍 JVM 类初始化的机制，然后给出几个易出错的实例来分析，帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段，分别是：loading 、Linking、initialization
阅读更多...
Spring Security 基于表达式的权限控制

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权，允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true （Spring security默认会带有ROLE_前缀）,去
阅读更多...
浅析Spring Security认证过程

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程，特意画了如下的类图，包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口，接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti
阅读更多...
Spring Security--Architecture Overview

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类，它们之间的依赖，构建起了整个框架。想要理解整个架构，最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文（security context）的信息。当前操作的用户是谁，该用户是否已经被认证，他拥有哪些角色权限…这些都被保
阅读更多...
Spring Security基于数据库验证流程详解

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
Spring Security 从入门到进阶系列教程

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
Java架构师知识体认识

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物
阅读更多...
Zookeeper安装和配置说明

Zookeeper安装和配置说明

一、Zookeeper的搭建方式 Zookeeper安装方式有三种，单机模式和集群模式以及伪集群模式。 ■ 单机模式：Zookeeper只运行在一台服务器上，适合测试环境； ■ 伪集群模式：就是在一台物理机上运行多个Zookeeper 实例； ■ 集群模式：Zookeeper运行于一个集群上，适合生产环境，这个计算机集群被称为一个“集合体”（ensemble） Zookeeper通过复制来实现
阅读更多...
中文分词jieba库的使用与实景应用（一）

中文分词jieba库的使用与实景应用（一）

知识星球：https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义： 精确模式（默认模式）： 全模式： 搜索引擎模式： paddle 模式（基于深度学习的分词模式）： 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2