准备
依赖
使用Curator，是目前使用者最多的zookeeper客户端
1
2
3
4
5
6
7
8
9
10
| <dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>5.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>5.2.0</version>
</dependency>
|
配置类
curator需要我们自己配置一个CuratorFramework
1
2
3
4
5
6
7
8
9
10
| @Configuration
public class ZookeeperConfig {
@Bean(initMethod = "start")
public CuratorFramework curatorFramework(){
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
return CuratorFrameworkFactory.newClient("localhost:2181", retryPolicy);
}
}
|
操作代码
实现类
zookeeper的每条命令curator都有单独的方法对应，需要设置的参数curator也都设置了单独的方法，比如set -v version，curator可以使用setData().setData().withVersion(version)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
| @Service
public class ZookeeperService {
private Logger logger = LoggerFactory.getLogger(ZookeeperService.class);
@Autowired
private CuratorFramework client;
public void creatNode(String path, String data) throws Exception {
client.create().withMode(CreateMode.PERSISTENT).forPath(path, data.getBytes(StandardCharsets.UTF_8));
}
public String getNode(String path) throws Exception {
byte[] data = client.getData().forPath(path);
return new String(data);
}
public void setNode(String path, String newData, int version) throws Exception {
client.setData().withVersion(version).forPath(path, newData.getBytes(StandardCharsets.UTF_8));
}
public void deleteNode(String path,int version) throws Exception {
client.delete().deletingChildrenIfNeeded().withVersion(version).forPath(path);
}
public void addWatch(String path) throws Exception {
client.watchers().add().withMode(AddWatchMode.PERSISTENT).usingWatcher((CuratorWatcher) watchedEvent -> logger.info("{}",watchedEvent)).forPath(path);
}
public void addWatch1(String path){
CuratorCache curatorCache = CuratorCache.build(client, path);
curatorCache.listenable().addListener((type, childData, childData1) -> {
switch (type){
case NODE_CREATED:
logger.info("创建节点：\n{}------{}",childData1.getPath(),new String(childData1.getData()));
break;
case NODE_CHANGED:
logger.info("节点改变：\n{}------{}\n{}------{}",childData.getPath(),new String(childData.getData()),childData1.getPath(),new String(childData1.getData()));
break;
case NODE_DELETED:
logger.info("节点删除：\n{}------{}",childData.getPath(),new String(childData.getData()));
break;
}
});
curatorCache.start();
logger.info("为{}添加watch成功",path);
}
}
|
Backgroundable接口
Backgroundable 接口中定义了多个方法，最常用的是 inBackground() 方法，它表示将当前的操作放在后台线程中执行，即异步执行。inBackground() 方法有多个重载形式，其中最简单的形式不需要传入任何参数，表示后台任务执行完成后，不需要回调任何方法。
如果需要在后台任务完成后执行回调方法，可以使用inBackground(org.apache.curator.framework.api.BackgroundCallback callback) 方法。该方法需要传入一个 BackgroundCallback 接口的实现对象，该接口中定义了 processResult() 方法，用于在后台任务执行完成后处理结果。
如果需要在后台任务完成后同时设置回调方法和自定义的 Executor，可以使用 inBackground(BackgroundCallback callback, java.util.concurrent.Executor executor) 方法。其中，Executor 对象表示后台任务执行所使用的线程池，通过它可以控制后台任务的执行。如果不指定 Executor，则会使用默认的线程池。
Curator 框架的许多操作都支持后台异步执行，可以通过 inBackground() 方法指定。在使用 Curator 进行开发时，可以根据实际情况选择同步或异步操作。如果需要执行一些比较耗时的操作，最好采用异步方式，避免阻塞应用程序的主线程。