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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
| @Component
public class ZkHelper {
//用于同时多个zk集群的使用和操作
private volatile Map<String, ZKClient> clientMap = Maps.newConcurrentMap();
@Value("${zookeeper.address}")
String connectionUrl;
@PreDestroy
public void closeZkClient() {
for (ZKClient zkClient : clientMap.values()) {
try {
zkClient.close();
} catch (Exception e) {
//...
}
}
}
/**
* 获取默认连接的zk对象
*/
public ZKClient getDefaultZKClient() {
ZKClient zkClient = getZKClient(connectionUrl);
return zkClient;
}
/**
* 根据连接地址来实例化并使用zk
*/
public ZKClient getZKClient(String address) {
Preconditions.checkArgument(StringUtils.isNotBlank(address), "address is illegal.");
if (clientMap.containsKey(address)) {
return clientMap.get(address);
}
synchronized (this) {
if (clientMap.containsKey(address)) {
return clientMap.get(address);
} else {
ZKClient zkClient = new ZKClient(address);
clientMap.put(address, zkClient);
return zkClient;
}
}
}
/**
* 具体对业务逻辑的封装
*/
public static class ZKClient {
private String connectString;
private CuratorFramework client;
private static final int TIMEOUT = 5000;
//注册,仅用作关闭!
private Map<String, NodeCache> nodeCacheMap = Maps.newConcurrentMap();
private Map<String, PathChildrenCache> pathChildrenCacheMap = Maps.newConcurrentMap();
public ZKClient(String address) {
this.connectString = address;
client = CuratorFrameworkFactory.newClient(address, TIMEOUT, TIMEOUT, new ExponentialBackoffRetry(TIMEOUT, 3));
client.start();
this.setConnWatcher(new ConnectionStateListener() {
@Override
public void stateChanged(CuratorFramework client, ConnectionState newState) {
//...
}
});
}
public CuratorFramework getZkCuratorClient() {
return client;
}
public boolean createNode(String path, CreateMode mode, byte[] data) {
try {
ZKPaths.mkdirs(client.getZookeeperClient().getZooKeeper(), path, false);
client.create().creatingParentsIfNeeded().withMode(mode).forPath(path, data);
return true;
} catch (Exception e) {
//...
}
return false;
}
public boolean createNode(String path, CreateMode mode) {
//...
}
public String getData(String path) {
//...
}
public boolean setData(String path, byte[] data) {
try {
if (!checkNodeExisted(path)) {
ZKPaths.mkdirs(client.getZookeeperClient().getZooKeeper(), path, true);
}
client.setData().forPath(path, data);
} catch (Exception e) {
return false;
}
return true;
}
/**
* 给指定路径节点设置一个监控,监控zk节点的变化,只监控一次
*/
public void setNodeWatcher(String path, CuratorWatcher curatorWatcher) {
Preconditions.checkArgument(checkNodeExisted(path), "the node does not exist.");
try {
client.getData().usingWatcher(curatorWatcher).forPath(path);
} catch (Exception e) {
//...
}
}
public boolean deleteNode(String path) {
if (!checkNodeExisted(path)) {
return false;
}
try {
client.delete().forPath(path);
} catch (Exception e) {
//...
return false;
}
return true;
}
/**
* 给指定路径节点的所有子节点设置一个监控,只监控一次
*/
public void setChildrenWatcher(String path, CuratorWatcher curatorWatcher) {
Preconditions.checkArgument(checkNodeExisted(path), "the node does not exist.");
try {
client.getChildren().usingWatcher(curatorWatcher).forPath(path);
} catch (Exception e) {
//...
}
}
public boolean checkNodeExisted(String path) {
if (StringUtils.isBlank(path)) {
return false;
}
Stat stat = null;
try {
stat = client.checkExists().forPath(path);
} catch (Exception e) {
//...
}
return stat == null ? false : true;
}
public boolean checkAndCreatePath(String path) {
if (checkNodeExisted(path)) {
return true;
}
try {
ZKPaths.mkdirs(client.getZookeeperClient().getZooKeeper(), path, true);
} catch (Exception e) {
//...
return false;
}
return true;
}
/**
* 给该连接设置一个监控,连接每发生一次变化,都会回调,重复监听
* Curator具有连接中断重连机制!
*/
public void setConnWatcher(ConnectionStateListener connWatcher) {
client.getConnectionStateListenable().addListener(connWatcher);
}
/**
* 监听Node节点中数据改变事件,自动重复监听
*/
public boolean setNodeCacheListener(String path, NodeCacheListener listener) {
if (!checkNodeExisted(path)) {
return false;
}
NodeCache nodeCache = new NodeCache(client, path);
nodeCacheMap.put(path, nodeCache);
try {
nodeCache.start();
nodeCache.getListenable().addListener(listener);
} catch (Exception e) {
return false;
}
return true;
}
/**
* 关闭指定path的NodeCache
*/
public void closeNodeCache(String path) {
if (nodeCacheMap.containsKey(path)) {
try {
nodeCacheMap.get(path).close();
} catch (IOException e) {
//...
}
}
}
public void closePathChildrenCache(String path) {
if (pathChildrenCacheMap.containsKey(path)) {
PathChildrenCache childrenCache = pathChildrenCacheMap.get(path);
closePathChildrenCache(childrenCache);
}
}
private void closePathChildrenCache(PathChildrenCache childrenCache) {
try {
childrenCache.close();
//清空缓存值
childrenCache.clear();
//清除map中的listener对象
childrenCache.getListenable().clear();
} catch (IOException e) {
//...
}
}
/**
* 关闭ZKClient对象
*/
public void close() {
for (NodeCache nodeCache : nodeCacheMap.values()) {
try {
nodeCache.close();
} catch (IOException e) {
//...
}
}
for (PathChildrenCache childrenCache : pathChildrenCacheMap.values()) {
closePathChildrenCache(childrenCache);
}
try {
client.close();
} catch (Exception e) {
//...
}
}
public boolean setPathChildListener(String path, PathChildrenCacheListener listener) {
if (!checkNodeExisted(path)) {
//...
return false;
}
PathChildrenCache childrenCache = new PathChildrenCache(this.getZkCuratorClient(), path, false);
pathChildrenCacheMap.put(path, childrenCache);
try {
childrenCache.start();
childrenCache.getListenable().addListener(listener);
} catch (Exception e) {
//...
return false;
}
return true;
}
}
}
|
