zookeeper应用案例之分布式服务器动态上下线感知
zookeeper应用案例之分布式服务器动态上下线感知,当服务器上线和下线时候客户端都能感知到,还有哪些机器在线.并对zookeeper管理的服务器进行节点的监听;代码实现:客户端
每当服务端有服务器上线或下线 在客户端都能通过监听感知到
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
package org.zookeeper.anli;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
public class ZookeeperClient {
private volatile List<String> servers = null;
private ZooKeeper zk = null;
// 获取zk连接
private void getZkClient() throws Exception {
zk = new ZooKeeper(GlobalConstants.zkhosts, GlobalConstants.sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
if(event.getType() == EventType.None) return;
try {
// 获取新的服务器列表,重新注册监听
updateServers();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
/**
* 从zk中获取在线服务器信息
*/
public void updateServers() throws Exception {
// 从servers父节点下获取到所有子节点，并注册监听
List<String> children = zk.getChildren(GlobalConstants.parentZnodePath, true);
ArrayList<String> serverList = new ArrayList<String>();
for (String child : children) {
byte[] data = zk.getData(GlobalConstants.parentZnodePath + "/" + child, false, null);
serverList.add(new String(data));
}
// 如果客户端是一个多线程程序，而且各个线程都会竞争访问servers列表，所以，在成员中用volatile修饰了一个servers变量
// 而在更新服务器信息的这个方法中，是用一个临时List变量来进行更新
servers = serverList;
// 将更新之后的服务器列表信息打印在控制台观察一下
for (String server : servers) {
System.out.println(server);
}
}
/**
* 业务逻辑
*
* @throws InterruptedException
*/
private void requestService() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
ZookeeperClient client = new ZookeeperClient();
// 先构造一个zk的连接
client.getZkClient();
// 获取服务器列表
client.updateServers();
// 客户端进入业务流程，请求服务器的服务
client.requestService();
}
}
代码实现:服务端
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
package org.zookeeper.anli;
import java.util.List;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
public class ZookeeperServer {
private ZooKeeper zk = null;
private volatile List<String> servers = null;
private void getZkClient() throws Exception{
//服务器在需求中并不需要做任何监听
zk = new ZooKeeper(GlobalConstants.zkhosts, GlobalConstants.sessionTimeout,null);
}
/**
* 向zookeeper中的/servers下创建子节点
* @throws InterruptedException
* @throws KeeperException
*/
private void connectZK(String serverName,String port) throws Exception{
//先创建出父节点
if(zk.exists(GlobalConstants.parentZnodePath, false) == null){
zk.create(GlobalConstants.parentZnodePath, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
//连接zk创建znode
zk.create(GlobalConstants.parentZnodePath+"/", (serverName+":"+port).getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("server " + serverName + " is online ......");
}
//服务器的具体业务处理功能
private void handle(String serverName) throws Exception {
System.out.println("server " + serverName + " is waiting for task process......");
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
ZookeeperServer server = new ZookeeperServer();
//获取与zookeeper通信的客户端连接
server.getZkClient();
//一启动就去zookeeper上注册服务器信息，参数1： 服务器的主机名 参数2：服务器的监听端口
//模拟 在servers节点下创建一个服务器 server01 1212 代表一台服务器上线
server.connectZK(args,args);
//进入业务逻辑处理流程
server.handle(args);
}
}
所需要用到的jar包:
测试方法,将server端打成一个jar包 client打成一个jar包 将server丢到服务器上 反复执行该jar包
在windows中执行client.jar包 查看服务器上下线感知
页:
[1]