1、监控ZOOKeeper的注册服务变化主要检测注册端口和节点的变化,服务端
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
public class ZkServerClient {
// 存放服务列表信息
public static List<String> listServer = new ArrayList<String>();
// 客户端:读取service节点,获取下面的子节点value值 本地实现远程调用。
private static ZkClient zkClient = new ZkClient("192.168.184.128:2181");
private static String parentService = "/service";
public static void main(String[] args) {
initServer();
ZkServerClient client = new ZkServerClient();
BufferedReader console = new BufferedReader(new InputStreamReader(System.in));
while (true) {
String name;
try {
name = console.readLine();
if ("exit".equals(name)) {
System.exit(0);
}
client.send(name);
} catch (IOException e) {
e.printStackTrace();
}
}
}
// 注册所有server
public static void initServer() {
// listServer.clear();
// listServer.add("127.0.0.1:8080");
// 从Zookeeper上获取服务列表信息
List<String> children = zkClient.getChildren(parentService);
getChildData(zkClient, children);
// 使用Zk事件通知获取最新服务列表信息
zkClient.subscribeChildChanges(parentService, new IZkChildListener() {
public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
System.out.println("注册中心服务里列表信息发生变化..");
getChildData(zkClient, currentChilds);
}
});
}
public static void getChildData(ZkClient zkClient, List<String> children) {
listServer.clear();
for (String p : children) {
String serverAddres = zkClient.readData(parentService + "/" + p);
listServer.add(serverAddres);
}
System.out.println("服务接口地址:" + listServer.toString());
}
// 请求总数
private static int reqCount = 1;
// 获取当前server信息
public static String getServer() {
int index = reqCount % listServer.size();
String addres = listServer.get(index);
System.out.println("客户端请求服务端:" + addres);
reqCount++;
return addres;
}
public void send(String name) {
String server = ZkServerClient.getServer();
String[] cfg = server.split(":");
Socket socket = null;
BufferedReader in = null;
PrintWriter out = null;
try {
socket = new Socket(cfg[0], Integer.parseInt(cfg[1]));
in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(), true);
out.println(name);
while (true) {
String resp = in.readLine();
if (resp == null)
break;
else if (resp.length() > 0) {
System.out.println("Receive : " + resp);
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
if (out != null) {
out.close();
}
if (in != null) {
try {
in.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
2、监控ZOOKeeper的注册服务变化主要检测注册端口和节点的变化,客户端来进行检测
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import org.I0Itec.zkclient.ZkClient;
//##ServerScoekt服务端
public class ZkServerScoekt implements Runnable {
private static int port = 8083;
private String parentService = "/service";
/**
* 服务器端:<br>
* 1.服务端启动的时候,会将当前服务信息注册到注册中心。首先先创建一个父节点为service,在父节点下面在创建一个子节点,
* 每个子节点都存放当前服务接口地址。<br>
* ##节点结构 <br>
* /service 持久节点<br>
* #####/8080 value 127.0.0.1:8080 临时节点<br>
* #####/8081 value 127.0.0.1:8081 临时节点<br>
*
*/
private ZkClient zkClient = new ZkClient("192.168.184.128:2181");
public static void main(String[] args) throws IOException {
ZkServerScoekt server = new ZkServerScoekt(port);
Thread thread = new Thread(server);
thread.start();
}
public ZkServerScoekt(int port) {
this.port = port;
}
private void regServer() {
// 1.先创建父节点service 为持久节点
if (!zkClient.exists(parentService)) {
// 2.创建父节点
zkClient.createPersistent(parentService);
}
String serverKey = parentService + "/server_" + port;
if (zkClient.exists(serverKey)) {
zkClient.delete(serverKey);
}
// 3.创建子节点 value为服务接口地址
zkClient.createEphemeral(serverKey, "192.168.184.128:" + port);
}
public void run() {
ServerSocket serverSocket = null;
try {
serverSocket = new ServerSocket(port);
System.out.println("Server start port:" + port);
regServer();
Socket socket = null;
while (true) {
socket = serverSocket.accept();
new Thread(new ServerHandler(socket)).start();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (serverSocket != null) {
serverSocket.close();
}
} catch (Exception e2) {
}
}
}
}