[Java知识库]Zookeeper服务器动态上下线监听

思路图示分析

Server.java

package com.lagou.zk.onoffline;

import org.I0Itec.zkclient.ZkClient;

//服务端提供client需要的要给时间查询服务，服务端向zk建立临时节点
public class Server {

    //创建ZK
    ZkClient zkClient = null;
    private void connectZk(){
        zkClient = new ZkClient("centos7-1:2181,centos7-2:2181");
        if (!zkClient.exists("/servers")){
            zkClient.createPersistent("/servers");
        }
    }

    //告知zk服务器相关信息
    private void saveServerInfo(String ip,String port){
        String sequencePath = zkClient.createEphemeralSequential("/servers/server", ip + ":" + port);
        System.out.println("服务器："+ip+":"+port+"向zk保存信息成功，成功上线可以接受client查询");
    }

    public static void main(String[] args) {
        //准备两个服务器启动上线（多线程模拟，一个线程代表一个服务器）
        Server server = new Server();
        server.connectZk();
        server.saveServerInfo(args[0],args[1]);

        //创建一个线程类，可以接受socket请求
        new TimeService(Integer.parseInt(args[1])).start();
    }

}

TimeService.java

package com.lagou.zk.onoffline;

import java.io.IOException;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Date;

public class TimeService extends Thread {

    private int port = 0;

    public TimeService(int port) {
        this.port = port;
    }

    @Override
    public void run() {
        //通过socket与client进行交流，启动serversocket监听请求
        try {
            //指定监听的端口
            ServerSocket serverSocket = new ServerSocket(port);

            while (true){
                Socket accept = serverSocket.accept();
                //server只发送一个时间值
                OutputStream out = accept.getOutputStream();
                out.write(new Date().toString().getBytes());
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

如下图所示，两个服务器已经起来


后台看一下是/servers否已经创建成功，可以看到有两个节点

Client.java

package com.lagou.zk.onoffline;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

//注册监听zk指定目录，维护自己本地一个servers信息，收到信息要进行更新
//发送时间查询请求并接受服务端返回的数据
public class Client {
    //创建ZK
    ZkClient zkClient = null;
    //创建一个servers信息集合
    ArrayList<String> infos = new ArrayList<>();

    private void connectZk(){
        zkClient = new ZkClient("centos7-1:2181,centos7-2:2181");
        //第一次获取服务器信息。所有的子节点
        List<String> childs = zkClient.getChildren("/servers");
        for (String child : childs) {
            Object o = zkClient.readData("/servers/" + child);
            infos.add(String.valueOf(o));
        }

        //对servers目录进行监听
        zkClient.subscribeChildChanges("/servers", new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> child) throws Exception {
                //接受到通知，说明节点发生了变化，client需要更新Infos集合中的数据
                ArrayList<String> list = new ArrayList<>();
                //便利更新后的节点
                for (String path : child) {
                    Object o = zkClient.readData("/servers/" + path);
                    list.add(String.valueOf(o));
                }
                infos = list;
                System.out.println("接收到通知，最新服务器信息为："+infos);
            }
        });
    }

    //发送时间查询的请求
    public void sendRequest() throws IOException {
        //目标服务器地址
        Random random = new Random();
        int i = random.nextInt(infos.size());
        String[] arr = infos.get(i).split(":");

        //建立socket连接
        Socket socket = new Socket(arr[0], Integer.parseInt(arr[1]));
        OutputStream out = socket.getOutputStream();
        InputStream in = socket.getInputStream();

        //发送数据
        out.write("query time".getBytes());
        byte[] b = new byte[1024];
        in.read(b);
        System.out.println("client端接受到server"+infos.get(i)+"返回结果"+new String(b));

        //释放资源
        in.close();
        out.close();
        socket.close();
    }

    public static void main(String[] args) throws IOException, InterruptedException {

        Client client = new Client();
        client.connectZk();
        while (true){
            //发送请求
            client.sendRequest();
            //每隔几秒发送一次请求到服务器
            Thread.sleep(2000);
        }
    }
}

接下来启动客户端 效果如下

已成功完成监听!