Blog
zookeeper-05 监听服务器节点动态上下线
- 需求
某分布式系统中,主节点可以多台,可以动态上下线,任意一台客户端都能实时感知到主节点服务器的上下线
- 分析
Server
public class DistributeServer {
public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
DistributeServer server = new DistributeServer();
// 创建zookeeper集群连接
server.getConnect();
// 注册服务器 (创建节点)
server.regist(args[0]);
// 业务逻辑 (不让进程退出)
server.business();
}
private void business() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
private void regist(String hostname) throws KeeperException, InterruptedException {
String path = zkClient.create("/servers/server", hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(hostname + "is online");
}
private String connectString = "hadoop001:2181,hadoop002:2181,hadoop003:2181";
private int sessionTimeout = 2000;
private ZooKeeper zkClient;
private void getConnect() throws IOException {
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
public void process(WatchedEvent event) {
}
});
}
}
Client
public class DistributeClient {
public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
DistributeClient server = new DistributeClient();
// 创建客户端连接
client.getConnect();
// 注册监听
client.getChildren();
// 业务逻辑
client.business();
}
private void business() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
private void getChildren() throws KeeperException, InterruptedException {
List<String> children = zkClient.getChildren("/servers", true);
// 存储服务器节点名称集合
ArrayList<String> hosts = new ArrayList<String>();
for (String child : children) {
byte[] data = zkClient.getData("/servers/" + child, false, null);
hosts.add(new String(data));
}
// 将所有在线主机名称打印到控制台
System.out.println(hosts);
}
private String connectString = "hadoop001:2181,hadoop002:2181,hadoop003:2181";
private int sessionTimeout = 2000;
private ZooKeeper zkClient;
private void getConnect() throws IOException {
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
public void process(WatchedEvent event) {
try {
// 监听响应执行下面函数
getChildren();
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}