介绍
springboot集成zookeeper，常用方式有2种：
- zookeeper官方
- Curator方式
本文介绍 curator方式
Curator
Curator是Apache开源的一个Zookeeper客户端连接和操作的组件，Curator框架在Zookeeper 原生API接口上进行二次包装。
提供ZooKeeper各种应用场景：比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等API封装。
创建springboot工程
https://www.malaoshi.top/show_1EF5qXjMmlZR.html
集成Curator
修改pom.xml
添加依赖
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-client</artifactId>
<version>2.12.0</version>
</dependency>
配置zookeeper
创建 application.yml，内容如下：
zoo:
keeper:
#开启标志
enabled: true
#服务器地址
server: 127.0.0.1:2181
#命名空间，被称为ZNode
namespace: aaa
#权限控制，加密
digest:
#会话超时时间
sessionTimeoutMs: 3000
#连接超时时间
connectionTimeoutMs: 60000
#最大重试次数
maxRetries: 2
#初始休眠时间
baseSleepTimeMs: 1000
读取配置类
@Component
@ConfigurationProperties(prefix = "zoo.keeper")
public class ZookeeperProperties {
private boolean enabled;
private String server;
private String namespace;
private int sessionTimeoutMs;
private int connectionTimeoutMs;
private int maxRetries;
private int baseSleepTimeMs;
private String digest;
public String getDigest() {
return digest;
}
public void setDigest(String digest) {
this.digest = digest;
}
public boolean isEnabled() {
return enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public String getServer() {
return server;
}
public void setServer(String server) {
this.server = server;
}
public String getNamespace() {
return namespace;
}
public void setNamespace(String namespace) {
this.namespace = namespace;
}
public int getSessionTimeoutMs() {
return sessionTimeoutMs;
}
public void setSessionTimeoutMs(int sessionTimeoutMs) {
this.sessionTimeoutMs = sessionTimeoutMs;
}
public int getConnectionTimeoutMs() {
return connectionTimeoutMs;
}
public void setConnectionTimeoutMs(int connectionTimeoutMs) {
this.connectionTimeoutMs = connectionTimeoutMs;
}
public int getMaxRetries() {
return maxRetries;
}
public void setMaxRetries(int maxRetries) {
this.maxRetries = maxRetries;
}
public int getBaseSleepTimeMs() {
return baseSleepTimeMs;
}
public void setBaseSleepTimeMs(int baseSleepTimeMs) {
this.baseSleepTimeMs = baseSleepTimeMs;
}
}
zookeeper配置类
连接zookeeper
@Configuration
public class ZookeeperConfig {
private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperConfig.class) ;
@Resource
private ZookeeperProperties zookeeperProperties;
private CuratorFramework client = null ;
/**
* 初始化
*/
@PostConstruct
public void init (){
//重试策略，初试时间1秒，重试10次
RetryPolicy policy = new ExponentialBackoffRetry(
zookeeperProperties.getBaseSleepTimeMs(),
zookeeperProperties.getMaxRetries());
//通过工厂创建Curator
client = CuratorFrameworkFactory.builder()
.connectString(zookeeperProperties.getServer())
.authorization("digest", zookeeperProperties.getDigest().getBytes())
.connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMs())
.sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMs())
.retryPolicy(policy).build();
//开启连接
client.start();
LOGGER.info("zookeeper 初始化完成...");
}
@Bean
public CuratorFramework getClient (){
return client ;
}
}
操作zookeeper类
@Component
public class ZookeeperClient {
private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperClient.class);
@Resource
CuratorFramework client = null;
/**
* 创建节点
* @param mode 节点类型
* 1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
* 2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
* 3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
* 4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
* @param path 节点名称
*/
public void createNode(CreateMode mode, String path) {
try {
// 递归创建所需父节点
client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
} catch (Exception e) {
LOGGER.error("createNode error...", e);
e.printStackTrace();
}
}
/**
* 创建节点
* @param mode 节点类型
* 1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
* 2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
* 3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
*4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
* @param path 节点名称
* @param nodeData 节点数据
*/
public void createNode(CreateMode mode, String path, String nodeData) {
try {
// 创建节点，关联数据
client.create().creatingParentsIfNeeded().withMode(mode)
.forPath(path,nodeData.getBytes("UTF-8"));
} catch (Exception e) {
LOGGER.error("createNode error...", e);
}
}
/**
* 设置指定节点的数据
* @param path
* @param nodeData
*/
public void setNodeData(String path, String nodeData) {
try {
client.setData().forPath(path, nodeData.getBytes("UTF-8"));
} catch (Exception e) {
LOGGER.error("setNodeData error...", e);
}
}
/**
* 获取数据时先同步
* @param path
* @return
*/
public String synNodeData(String path){
client.sync();
return getNodeData( path);
}
/**
* 获取指定节点的数据
* @param path
* @return
*/
public String getNodeData(String path) {
try {
// 数据读取和转换
byte[] dataByte = client.getData().forPath(path) ;
String data = new String(dataByte,"UTF-8") ;
return data ;
}catch (Exception e) {
LOGGER.error("getNodeData error...", e);
}
return null;
}
public List<String> getNodeChild(String path) {
List<String> nodeChildDataList = new ArrayList<>();
try {
// 节点下数据集
nodeChildDataList = client.getChildren().forPath(path);
} catch (Exception e) {
LOGGER.error("getNodeChild error...", e);
}
return nodeChildDataList;
}
/**
* 删除节点数据
* @param path
* @param deleteChildre 是否删除子节点
*/
public void deleteNode(String path, Boolean deleteChildre) {
try {
if(deleteChildre) {
// 递归删除节点
client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
} else {
// 删除单个节点
client.delete().guaranteed().forPath(path);
}
} catch (Exception e) {
LOGGER.error("deleteNode error...", e);
}
}
public void closeClient (){
if (client != null){
client.close();
}
}
public boolean isExistNode(String path) {
client.sync() ;
try {
return client.checkExists().forPath(path) != null;
} catch (Exception e) {
LOGGER.error("isExistNode error...", e);
}
return false;
}
/**
* 监听数据节点的变化情况
* @param watchPath
* @param listener
*/
public void watchPath(String watchPath, TreeCacheListener listener) {
// NodeCache nodeCache = new NodeCache(client, watchPath, false);
TreeCache cache = new TreeCache(client, watchPath);
cache.getListenable().addListener(listener);
try {
cache.start();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 获取读写锁
* @param path
* @return
*/
public InterProcessReadWriteLock getReadWriteLock(String path) {
// 写锁互斥、读写互斥
InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, path);
return readWriteLock ;
}
}
测试Controller
@RestController
public class TestCtrl {
@Resource
private ZookeeperClient zkClient;
/*
http://localhost:8081/create?path=/service/notice&data=http://192.168.1.100:8080
*/
@RequestMapping("/create")
public String createNode(String path,String data){
zkClient.createNode(CreateMode.PERSISTENT,path,data);
return "创建成功！";
}
/*
http://localhost:8081/getNodeData?path=/service/notice
*/
@RequestMapping("/getNodeData")
public String getNodeData(String path){
return zkClient.getNodeData(path);
}
/*
http://localhost:8081/set?path=/service/notice&data=http://192.168.1.200:8080
*/
@RequestMapping("/set")
public String setNodeData(String path,String data){
zkClient.setNodeData(path,data);
return "修改成功！";
}
/*
http://localhost:8081/getNodeChild?path=/
*/
@RequestMapping("/getNodeChild")
public List<String> getNodeChild(String path){
return zkClient.getNodeChild(path);
}
/*
http://localhost:8081/deleteNode?path=/service
*/
@RequestMapping("/deleteNode")
public String deleteNode(String path){
zkClient.deleteNode(path,true);
return "删除成功！";
}
/*
http://localhost:8081/isExistNode?path=/
*/
@RequestMapping("/isExistNode")
public boolean isExistNode(String path){
return zkClient.isExistNode(path);
}
/*
说明见：https://www.malaoshi.top/show_1IXKll9pCf8.html
访问http://localhost:8081/watchPath
然后在客户端执行下面命令：
create /listner/node-1
set /listener/node-2 http://192.168.1.222:8080
create /listner/node-2 http://192.168.1.100:8080
delete /listener/node-2
key监听多次操作
*/
@RequestMapping("/watchPath")
public String watchPath() throws InterruptedException {
//监听路径
String workerPath = "/listener";
boolean exist=zkClient.isExistNode(workerPath);
if(!exist) {
zkClient.createNode(CreateMode.PERSISTENT, workerPath);
}
//子节点路径
String subWorkerPath = "/listener/node-";
TreeCacheListener listener=new TreeCacheListener(){
@Override
public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
TreeCacheEvent.Type type=treeCacheEvent.getType();
ChildData cd = treeCacheEvent.getData();
if(type==TreeCacheEvent.Type.NODE_ADDED) {
String path=cd.getPath();
byte[] dataByte=cd.getData();
if(dataByte!=null) {
String data = new String(dataByte, "utf-8");
System.out.println(String.format("添加子节点：%s - %s ", path,data));
}else{
System.out.println(String.format("添加子节点：%s ", path));
}
}else if(type==TreeCacheEvent.Type.NODE_REMOVED){
String path=cd.getPath();
byte[] dataByte=cd.getData();
if(dataByte!=null) {
String data = new String(dataByte, "utf-8");
System.out.println(String.format("删除子节点：%s - %s ", path,data));
}else{
System.out.println(String.format("删除子节点：%s ", path));
}
}else if(type==TreeCacheEvent.Type.NODE_UPDATED){
String path=cd.getPath();
byte[] dataByte=cd.getData();
if(dataByte!=null) {
String data = new String(dataByte, "utf-8");
System.out.println(String.format("更新节点：%s - %s ", path,data));
}else{
System.out.println(String.format("更新节点：%s ", path));
}
}
}
};
zkClient.watchPath(workerPath,listener);
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// for (int i = 0; i < 3; i++) {
// zkClient.createNode(CreateMode.PERSISTENT,subWorkerPath + i);
// }
//
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// for (int i = 0; i < 2; i++) {
// zkClient.deleteNode(subWorkerPath + i,true);
// }
Thread.sleep(Integer.MAX_VALUE);
return "监听成功！";
}
}
监听说明见：https://www.malaoshi.top/show_1IXKll9pCf8.html