第 5 篇 : SpringBoot整合Zookeeper框架curator
1. 在redis项目中,增加curator依赖,刷新maven<!-- Netflix的curator-recipes --><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version
·
1. 在redis项目中,增加curator依赖,刷新maven
<!-- Netflix的curator-recipes -->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>5.2.0</version>
</dependency>
2. 在application.yml文件中添加curator配置
zookeeper:
connectString: 192.168.109.160:2181,192.168.109.161:2181,192.168.109.162:2181
sessionTimeoutMs: 50000
baseSleepTimeMs: 1000
maxRetries: 3
connectionTimeoutMs: 50000
3. ZookeeperConfig,ZookeeperEnum,ZookeeperRequest
package com.hahashou.test.redis.config;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @description: {@link CuratorFrameworkFactory}的builder模式(或者通过 .newClient(...) 静态方法)
* @author: 哼唧兽
* @date: 9999/9/21
**/
@Configuration
@Data
@Slf4j
public class ZookeeperConfig {
/** 集群地址 */
@Value("${zookeeper.connectString}")
private String connectString;
/** 连接超时时间 */
@Value("${zookeeper.connectionTimeoutMs}")
private Integer connectionTimeoutMs;
/** 会话超时时间 */
@Value("${zookeeper.sessionTimeoutMs}")
private Integer sessionTimeoutMs;
/** 重试间隔等待的初试时间 */
@Value("${zookeeper.baseSleepTimeMs}")
private Integer baseSleepTimeMs;
/** 最大重试次数 */
@Value("${zookeeper.maxRetries}")
private Integer maxRetries;
/** 预定义的命名空间 */
@Value("${zookeeper.namespace}")
private String namespace;
@Bean
public CuratorFramework curatorFramework() {
CuratorFramework curatorFramework = CuratorFrameworkFactory
.builder()
.connectString(connectString)
.connectionTimeoutMs(connectionTimeoutMs)
.sessionTimeoutMs(sessionTimeoutMs)
.retryPolicy(new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries))
.namespace(namespace)
.build();
curatorFramework.start();
return curatorFramework;
}
}
package com.hahashou.test.redis.enums;
import lombok.Getter;
/**
* @description: 涉及Zookeeper的一些静态值
* @author: 哼唧兽
* @date: 9999/9/21
**/
public enum ZookeeperEnum {
ALL_AUTH("world", "anyone"),
SYMBOL("/"),
NAMESPACE("hahashou"),
;
@Getter
private String first;
@Getter
private String second;
ZookeeperEnum(String first) {
this.first = first;
}
ZookeeperEnum(String first, String second) {
this.first = first;
this.second = second;
}
}
package com.hahashou.test.redis.request;
import lombok.Data;
/**
* @description: 入参
* @author: 哼唧兽
* @date: 9999/9/21
**/
@Data
public class ZookeeperRequest {
/** 路径(需带/) */
private String path;
/** 节点数据 */
private String data;
/** 是否递归删除节点(当节点不为空时,不可以是false) */
private Boolean recursive;
}
4. TestZookeeperController,并启动项目
package com.hahashou.test.redis.controller;
import com.hahashou.test.redis.enums.ZookeeperEnum;
import com.hahashou.test.redis.request.ZookeeperRequest;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
/**
* @description: 测试Zookeeper
* @author: 哼唧兽
* @date: 9999/9/21
**/
@RestController
@RequestMapping("/zookeeper")
@Api(tags = "测试Zookeeper")
@Slf4j
public class TestZookeeperController {
@Resource
private CuratorFramework curatorFramework;
@PostMapping("/createNode")
@ApiOperation(value = "创建节点")
public String createNode(@RequestBody ZookeeperRequest zookeeperRequest) {
String path = zookeeperRequest.getPath(),
data = zookeeperRequest.getData();
ZookeeperEnum allAuth = ZookeeperEnum.ALL_AUTH;
List<ACL> aclList = Arrays.asList(new ACL(ZooDefs.Perms.ALL, new Id(allAuth.getFirst(), allAuth.getSecond())));
try {
validPath(path);
curatorFramework.create()
//没有父节点时 创建父节点
.creatingParentsIfNeeded()
//节点类型
.withMode(CreateMode.PERSISTENT)
// 配置权限
.withACL(aclList)
.forPath(path, data.getBytes());
} catch (Exception e) {
return "节点创建失败 : " + e.getMessage();
}
return "节点创建成功";
}
@PostMapping("/queryNode")
@ApiOperation(value = "查询节点数据")
public String queryNode(@RequestBody ZookeeperRequest zookeeperRequest) {
String path = zookeeperRequest.getPath();
String namespace = "/" + ZookeeperEnum.NAMESPACE.getFirst();
try {
validPath(path);
Stat stat = curatorFramework.checkExists().forPath(path);
if (stat == null) {
return "不存在该节点";
}
String dataString = new String(curatorFramework.getData().forPath(path));
return "节点 " + namespace + path + " 的数据为 : " + dataString;
} catch (Exception e) {
log.error("查询节点失败 : {}", e.getMessage());
return "";
}
}
@PostMapping("/queryChildNodes")
@ApiOperation(value = "查询子节点及其数据")
public List<ZookeeperRequest> queryChildNodes(@RequestBody ZookeeperRequest zookeeperRequest) {
List<ZookeeperRequest> result = new ArrayList<>();
String path = zookeeperRequest.getPath();
String symbol = ZookeeperEnum.SYMBOL.getFirst();
if (StringUtils.isEmpty(path)) {
path = symbol;
}
try {
List<String> childList = curatorFramework.getChildren().forPath(path);
for (String child : childList) {
ZookeeperRequest zookeeperResponse = new ZookeeperRequest();
boolean root = path.equals(symbol);
child = root ? child : symbol + child;
zookeeperResponse.setPath(root ? symbol + child : child);
zookeeperResponse.setData(new String(curatorFramework.getData().forPath(path+child)));
result.add(zookeeperResponse);
}
} catch (Exception e) {
log.error("查询节点失败 : {}", e.getMessage());
}
return result;
}
public void validPath(String path) throws Exception {
if (StringUtils.isEmpty(path)) {
throw new Exception("path为null或空");
} else if (!path.startsWith(ZookeeperEnum.SYMBOL.getFirst())) {
throw new Exception("路径没有带/");
}
}
@PostMapping("/updateNodeData")
@ApiOperation(value = "更新节点数据")
public String updateNodeData(@RequestBody ZookeeperRequest zookeeperRequest) {
String path = zookeeperRequest.getPath(),
data = zookeeperRequest.getData();
try {
validPath(path);
//也可指定版本更新,只有和节点版本一致才可更新成功 *.setData().withVersion(version).forPath(...);
Stat stat = curatorFramework.setData().forPath(path, data.getBytes());
log.info("stat : {}", stat);
} catch (Exception exception) {
log.error("更新数据失败 : {}", exception.getMessage());
return "更新失败";
}
return "更新成功";
}
@PostMapping("/deleteNode")
@ApiOperation(value = "删除节点")
public String deleteNode(@RequestBody ZookeeperRequest zookeeperRequest) {
String path = zookeeperRequest.getPath();
Boolean recursive = zookeeperRequest.getRecursive();
try {
validPath(path);
if (recursive) {
curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
} else {
curatorFramework.delete().forPath(path);
}
} catch (Exception exception) {
log.error("删除数据失败 : {}", exception.getMessage());
return "删除失败";
}
return "删除成功";
}
}
5. 测试
5.1 依次新增7个节点
{"data": "100","path": "/first"}
{"data": "firstA","path": "/first/a"}
{"data": "firstB","path": "/first/b"}
{"data": "firstC","path": "/first/c"}
{"data": "200","path": "/second"}
{"data": "secondD","path": "/second/d"}
{"data": "secondF","path": "/second/f"}
5.2 查询节点数据
{"path": "/first"}
5.3 查询子节点及其数据
{"path": "/"}
{"path": "/first"}
5.4 更新节点数据
{"data": "测试更新","path": "/first/a"}
再次查询
{"path": "/first/a"}
5.5 删除节点
递归删除 /second 节点
{"path": "/second","recursive": true}
非递归删除 /first/c 节点
{"path": "/first/c","recursive": false}
进入zookeeper客户端
ls /hahashou
ls /hahashou/first
权威｜前沿｜技术｜干货｜国内首个API全生命周期开发者社区
更多推荐
0
0- 0
已为社区贡献1条内容
所有评论(0)