Java Spring Boot 集成Zookeeper
更新时间：2022年08月18日 09:49:25 作者：剑圣无痕
这篇文章主要介绍了Java Spring Boot 集成Zookeeper，文章围绕主题展开详细的内容介绍，具有一定的参考价值，需要的小伙伴可以参考一下
集成步骤
1.pom.xml文件配置，引入相关jar包
Curator是Netflix公司开源的一套zookeeper客户端框架，解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册Watcher和NodeExistsException异常等等。
<!-- 封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式Barrier -->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.10.0</version>
<exclusions>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.13</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>特殊说明： 1.无需引入curator-framework，因为curator-recipes自动关联依赖引入curator-framework。 2.curator会默认引入zookeeper的jar报，需要检查版本与服务器的版本是否一致，如果不一致则需要排除引入 3.
2. 核心配置类
@Configuration
public class ZookeeperConfig implements Serializable
{
private static final long serialVersionUID = -9025878246972668136L;
private final ZooKeeperProperty zooKeeperProperty;
public ZookeeperConfig(ZooKeeperProperty zooKeeperProperty) {
this.zooKeeperProperty = zooKeeperProperty;
}
@Bean
public CuratorFramework curatorFramework()
{
RetryPolicy retryPolicy = new ExponentialBackoffRetry(zooKeeperProperty.getBaseSleepTime(),
zooKeeperProperty.getMaxRetries());
CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString(zooKeeperProperty.getServers())
.connectionTimeoutMs(zooKeeperProperty.getConnectionTimeout())
.sessionTimeoutMs(zooKeeperProperty.getSessionTimeout())
.retryPolicy(retryPolicy)
.build();
client.start();
return client;
}
@Bean
@ConditionalOnMissingBean
public ZooKeeperUtils zooKeeperTemplate(CuratorFramework client) {
return new ZooKeeperUtils(client);
}
}
@ConfigurationProperties(prefix="zookeeper")
@Component
public class ZooKeeperProperty implements Serializable
{
private static final long serialVersionUID = 8650758711482699256L;
/**
* zk连接集群，多个用逗号隔开
*/
private String servers;
/**
* 会话超时时间
*/
private int sessionTimeout = 60000;
/**
* 连接超时时间
*/
private int connectionTimeout = 15000;
/**
* 初始重试等待时间(毫秒)
*/
private int baseSleepTime = 1000;
/**
* 重试最大次数
*/
private int maxRetries = 10;
//省略get、set方法
......
}3.常用API功能
@Component
public class ZooKeeperUtils
{
private static final Logger logger = LoggerFactory
.getLogger(ZooKeeperUtils.class);
/**
* 路径分隔符
*/
private static final String PATH_SEPARATOR = "/";
/**
* zk连接
*/
private final CuratorFramework client;
public ZooKeeperUtils(CuratorFramework client)
{
this.client = client;
}
/**
* 创建空节点，默认持久节点
*
* @param path
* 节点路径
* @param node
* 节点名称
* @return 完整路径
*/
public String createNode(String path, String node)
{
return createNode(path, node, CreateMode.PERSISTENT);
}
/**
* 创建带类型的空节点
*
* @param path
* 节点路径
* @param node
* 节点名称
* @param createMode
* 类型 CreateMode.PERSISTENT: 创建节点后，不删除就永久存在
* CreateMode.PERSISTENT_SEQUENTIAL：节点path末尾会追加一个10位数的单调递增的序列
* CreateMode.EPHEMERAL：创建后，回话结束节点会自动删除
* CreateMode.EPHEMERAL_SEQUENTIAL：节点path末尾会追加一个10位数的单调递增的序列
* @return 路径
*/
public String createNode(String path, String node, CreateMode createMode)
{
path = buildPath(path, node);
logger.info("create node for path: {} with createMode: {}", path,
createMode.name());
try
{
client.create().creatingParentsIfNeeded().withMode(createMode)
.forPath(path);
logger.info("create node :{} sucessfully", node);
return path;
}
catch (Exception e)
{
logger.error(
"create node for path: {} with createMode: {} failed!",
path, createMode.name(), e);
return null;
}
}
/**
* 创建节点，默认持久节点
*
* @param path
* 节点路径
* @param node
* 节点名称
* @param value
* 节点值
* @return 完整路径
*/
public String createNode(String path, String node, String value)
{
return createNode(path, node, value, CreateMode.PERSISTENT);
}
/**
* 创建节点，默认持久节点
*
* @param path
* 节点路径
* @param node
* 节点名称
* @param value
* 节点值
* @param createMode
* 节点类型
* @return 完整路径
*/
public String createNode(String path, String node, String value,
CreateMode createMode)
{
if (Objects.isNull(value))
{
logger.error("ZooKeeper节点值不能为空!");
}
path = buildPath(path, node);
logger.info("create node for path: {}, value: {}, with createMode: {}",
path, value, createMode.name());
try
{
client.create().creatingParentsIfNeeded().withMode(createMode)
.forPath(path, value.getBytes());
return path;
}
catch (Exception e)
{
logger.error(
"create node for path: {}, value: {}, with createMode: {} failed!",
path, value, createMode.name(), e);
}
return null;
}
/**
* 获取节点数据
*
* @param path
* 路径
* @param node
* 节点名称
* @return 完整路径
*/
public String get(String path, String node)
{
path = buildPath(path, node);
try
{
byte[] bytes = client.getData().forPath(path);
if (bytes.length > 0)
{
return new String(bytes);
}
}
catch (Exception e)
{
logger.error("get value for path: {}, node: {} failed!", path,
node, e);
}
return null;
}
/**
* 更新节点数据
*
* @param path
* 节点路径
* @param node
* 节点名称
* @param value
* 更新值
* @return 完整路径
*/
public String update(String path, String node, String value)
{
if (Objects.isNull(value))
{
logger.error("ZooKeeper节点值不能为空!");
}
path = buildPath(path, node);
logger.info("update path: {} to value: {}", path, value);
try
{
client.setData().forPath(path, value.getBytes());
return path;
}
catch (Exception e)
{
logger.error("update path: {} to value: {} failed!", path, value);
}
return null;
}
/**
* 删除节点，并且递归删除子节点
*
* @param path
* 路径
* @param node
* 节点名称
* @return 路径
*/
public boolean delete(String path, String node)
{
path = buildPath(path, node);
logger.info("delete node for path: {}", path);
try
{
client.delete().deletingChildrenIfNeeded().forPath(path);
return true;
}
catch (Exception e)
{
logger.error("delete node for path: {} failed!", path);
}
return false;
}
/**
* 获取子节点
*
* @param path
* 节点路径
* @return
*/
public List<String> getChildren(String path)
{
if (StringUtils.isEmpty(path))
{
return null;
}
if (!path.startsWith(PATH_SEPARATOR))
{
path = PATH_SEPARATOR + path;
}
try
{
return client.getChildren().forPath(path);
}
catch (Exception e)
{
logger.error("get children path:{} error", path, e);
}
return null;
}
/**
* 判断节点是否存在
*
* @param path
* 路径
* @param node
* 节点名称
* @return 结果
*/
public boolean exists(String path, String node)
{
try
{
List<String> list = getChildren(path);
return !CollectionUtils.isEmpty(list) && list.contains(node);
}
catch (Exception e)
{
return false;
}
}
/**
* 申请锁，指定请求等待时间
*
* @param path
* 加锁zk节点
* @param time
* 时间
* @param unit
* 时间单位
* @param runnable
* 执行方法
*/
public void lock(String path, long time, TimeUnit unit, Runnable runnable)
{
try
{
InterProcessMutex lock = new InterProcessMutex(client, path);
if (lock.acquire(time, unit))
{
try
{
runnable.run();
}
finally
{
lock.release();
}
}
else
{
logger.error("获取锁超时：{}!", path);
}
}
catch (Exception e)
{
logger.error("获取锁失败: {}!", path);
}
}
/**
* 申请锁，指定请求等待时间
*
* @param path
* 加锁zk节点
* @param time
* 时间
* @param unit
* 时间单位
* @param callable
* 执行方法
* @return .
*/
public <T> T lock(String path, long time, TimeUnit unit,
Callable<T> callable)
{
try
{
InterProcessMutex lock = new InterProcessMutex(client, path);
if (lock.acquire(time, unit))
{
try
{
return callable.call();
}
finally
{
lock.release();
}
}
else
{
logger.error("获取锁超时：{}!", path);
}
}
catch (Exception e)
{
logger.error("获取锁失败: {}!", path);
}
return null;
}
/* *//**
* 对一个节点进行监听，监听事件包括指定的路径节点的增、删、改的操作
*
* @param path
* 节点路径
* @param listener
* 回调方法
* @throws Exception
*/
public void watchNode(String path,boolean dataIsCompressed,final ZooKeeperCallback zooKeeperCallback)throws Exception
{
try
{
final NodeCache nodeCache = new NodeCache(client, path,dataIsCompressed);
nodeCache.getListenable().addListener(new NodeCacheListener()
{
public void nodeChanged() throws Exception
{
ChildData childData = nodeCache.getCurrentData();
logger.info("ZNode节点状态改变, path={}", childData.getPath());
logger.info("ZNode节点状态改变, data={}", childData.getData());
logger.info("ZNode节点状态改变, stat={}", childData.getStat());
//处理业务逻辑
zooKeeperCallback.call();
}
});
nodeCache.start();
}
catch (Exception e)
{
logger.error("创建NodeCache监听失败, path={}",path);
}
}
/**
* 对指定的路径节点的一级子目录进行监听，不对该节点的操作进行监听，对其子目录的节点进行增、删、改的操作监听
*
* @param path
* 节点路径
* @param listener
* 回调方法
*/
public void watchChildren(String path, PathChildrenCacheListener listener)
{
try
{
PathChildrenCache pathChildrenCache = new PathChildrenCache(client,
path, true);
pathChildrenCache.start(PathChildrenCache.StartMode.NORMAL);
pathChildrenCache.getListenable().addListener(listener);
}
catch (Exception e)
{
logger.error("watch children failed for path: {}", path, e);
}
}
/**
* 将指定的路径节点作为根节点（祖先节点），对其所有的子节点操作进行监听，呈现树形目录的监听，可以设置监听深度，最大监听深度为2147483647（
* int类型的最大值）
*
* @param path
* 节点路径
* @param maxDepth
* 回调方法
* @param listener
* 监听
*/
public void watchTree(String path, int maxDepth, TreeCacheListener listener)
{
try
{
TreeCache treeCache = TreeCache.newBuilder(client, path)
.setMaxDepth(maxDepth).build();
treeCache.start();
treeCache.getListenable().addListener(listener);
}
catch (Exception e)
{
logger.error("watch tree failed for path: {}", path, e);
}
}
public String buildPath(String path, String node)
{
if (StringUtils.isEmpty(path) || StringUtils.isEmpty(node))
{
logger.error("ZooKeeper路径或者节点名称不能为空！");
}
if (!path.startsWith(PATH_SEPARATOR))
{
path = PATH_SEPARATOR + path;
}
if (PATH_SEPARATOR.equals(path))
{
return path + node;
}
else
{
return path + PATH_SEPARATOR + node;
}
}
}4.基本使用
@Autowired
private ZooKeeperUtils zooKeeperUtil;
@RequestMapping("/addNode")
public String addNode()
{
String path= zooKeeperUtil.createNode("/zookeeper", "node1");
return path;
}特殊说明：关于zookeeper的分布式锁，后续讲解常用分布式锁的时候，会详细说明。
常见错误和解决办法
问题1：调用api创建zookeeper节点时，报KeeperErrorCode = Unimplemented for /test错误。
原因：服务器安装zookeeper的版本与程序中的zookeeper版本不一致。
解决方案： 登录服务器，查看zookeeper安装版本，执行如下命令：
echo stat|nc 127.0.0.1 2181
当前引入的zookeeper版本为3.4.13，而zookeeper的版本与curator对应关系如下：
Curator 2.x.x - compatible with both ZooKeeper 3.4.x and ZooKeeper 3.5.x Curator 4.x.x - compatible only with ZooKeeper 3.5.x and includes support for new features such as dynamic reconfiguration, etc. Curator 5.x.x compatible only with ZooKeeper 3.6.x+
问题2：启动项目的日志中会有Will not attempt to authenticate using SASL错误
起初我认为是zookeeper需要进行SASL认证，但是通过查阅相关资料后，才知道3.4之前版本，zookeeper默认会采用SASL认证,3.4以后的版本没有此类问题。
到此这篇关于Java Spring Boot 集成Zookeeper的文章就介绍到这了,更多相关Spring Boot 集成Zookeeper内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了Java8 中的并行流 ParallelStreams，Java8并行流ParallelStream和Stream的区别就是支持并行执行，提高程序运行效率。下面就来看看文章内容具体介绍吧2021-10-10
这篇文章主要给大家介绍了关于spring参数传递与检验的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作能带来一定的帮助，需要的朋友们下面跟着小编来一起学习学习吧。2017-07-07
Prime算法是一种穷举查找算法来从一个连通图中构造一棵最小生成树。本文主要为大家介绍了Java中Prime算法的原理与实现，感兴趣的可以学习一下2022-07-07
通义千问是阿里云自主研发的大语言模型,能够在用户自然语言输入的基础上,通过自然语言理解和语义分析,理解用户意图,在不同领域、任务内为用户提供服务和帮助,下面这篇文章主要给大家介绍了关于java调用通义千问API的详细完整步骤,需要的朋友可以参考下2024-02-02
这篇文章主要介绍了大数组元素差异removeAll与Map效率对比,有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2023-03-03
这篇文章主要给大家介绍了关于JAVA最容易忽视的数据类型之枚举的相关资料,Java中的枚举类型是一种特殊的类型,它允许程序员定义一个固定的值集合,并为每个值分配一个名称,枚举类型提供了一种简单、安全和可读性强的方式来表示一组相关的常量,需要的朋友可以参考下2023-10-10
Volatile修饰的成员变量在每次被线程访问时，都强迫从共享内存中重读该成员变量的值。而且，当成员变量发生变化时，强迫线程将变化值回写到共享内存2013-08-08
当我们使用mybatis的时候,可能经常会碰到一批数据的批量更新问题,因为如果一条数据一更新,那每一条数据就需要涉及到一次数据库的操作,本文主要介绍了mybatis中批量更新多个字段的2种实现方法,感兴趣的可以了解一下2023-09-09
这篇文章主要介绍了Java爬虫爬取漫画示例,大部分的爬虫入门教学都是爬取图片的,本文就来测试一下爬取网站的漫画,需要的朋友可以参考下2023-04-04
这篇文章主要介绍了Java正则提取中括号中的内容操作,涉及Java针对字符串的正则匹配、转换、遍历等相关操作技巧,需要的朋友可以参考下2018-06-06
最新评论