以服务动态上下线通知为例
Client 监听服务器状态
public class DistributeClient {
private String connectString = "127.0.0.1:2181";
private int sessionTimeout = 2000;
private ZooKeeper zkClient;
public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
BasicConfigurator.configure();
DistributeClient client = new DistributeClient();
// 获取 zookeeper 集群连接
client.getConnect();
// 注册监听
client.getChlidren();
Thread.sleep(Long.MAX_VALUE);
}
private void getConnect() throws IOException {
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
try {
// 具体监听业务
getChlidren();
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
private void getChlidren() throws KeeperException, InterruptedException {
if (zkClient.exists("/servers", false) == null) {
zkClient.create("/servers", "server".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
// 监听 /servers 节点
List<String> children = zkClient.getChildren("/servers", true);
// 存储服务器节点主机名称集合
ArrayList<String> hosts = new ArrayList<>();
for (String child : children) {
// 获取节点内容，即主机名称
byte[] data = zkClient.getData("/servers/" + child, false, null);
hosts.add(new String(data));
}
System.out.println("在线主机：" + hosts);
}
}
Server 服务器，上线后 Client 端会收到通知
public class DistributeServer {
private String connectString = "127.0.0.1:2181";
private int sessionTimeout = 2000;
private ZooKeeper zkClient;
public static void main(String[] args) throws Exception {
BasicConfigurator.configure();
DistributeServer server = new DistributeServer();
// 连接 zookeeper 集群
server.getConnect();
// 注册节点
server.regist(UUID.randomUUID().toString());
Thread.sleep(Long.MAX_VALUE);
}
private void getConnect() throws IOException {
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
}
});
}
private void regist(String hostname) throws KeeperException, InterruptedException {
// 创建临时带序号节点
zkClient.create("/servers/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(hostname + "：上线");
}
}
测试
1.直接运行 Client
2.运行 Server 后再查看 Client 的控制台
3.关闭 Server 后再查看 Client 的控制台
