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内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
非常的实现原理，代码较多，建议大家仔细看看。2008-10-10
MapStruct是一个Java注解处理器框架,用于简化Java Bean之间的映射,本文主要介绍了Java中MapStruct入门使用及对比,感兴趣的可以了解一下2023-12-12
Java String字符串和Unicode字符相互转换代码详解
这篇文章主要介绍了Java String字符串和Unicode字符相互转换代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2019-05-05
这篇文章主要介绍了Java之线程池使用与原理全面解析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-03-03
这篇文章主要为大家介绍了Java语言中的文件数据流示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2022-11-11
这篇文章主要介绍了Java类锁、对象锁、私有锁冲突测试,得出结论是加锁方法够成了竞争关系,同一时刻只能有一个方法能执行,需要的朋友可以参考下2014-10-10
这篇文章主要为大家介绍了Spring Data JPA命名约定查询实现方法示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2023-12-12
这篇文章主要介绍了简单探索 Java 中的惰性计算，惰性计算（尽可能延迟表达式求值）是许多函数式编程语言的特性。惰性集合在需要时提供其元素，无需预先计算它们，这带来了一些好处。,需要的朋友可以参考下2019-06-06
这篇文章主要介绍了解决IDEA JSP没有代码提示问题的几种方法，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2019-10-10
IntelliJ IDEA 2020.2 全家桶及以下版本激活工具大全【喜讯】
这篇文章主要介绍了IntelliJ IDEA 2020.2 全家桶及以下版本激活工具大全【喜讯】,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2020-09-09
最新评论