Zookeeper實現負載均衡

在Nginx和SpringCloud中都可以實現負載均衡，在Zookeeper中也可以實現負載均衡

 

Zookeeper實現負載均衡：

 

 

 

生產者集群，創建一個/path的父節點這個節點是持久節點，集群中的每個生產者分別在父節點中創建子節點(示例根據端口創建) 生產者一創建節點/path/producer1 生產者二創建節點/path/producer2|(節點的值都是ip地址和端口),該節點是臨時節點，zookeeper斷開連接則該節點消失。消費者可以通過獲取zookeeper中/path父節點下面的所有子節點信息。然后根據自己的需求在本地實現負載均衡策略(ip綁定，權重，輪訓)

 

java 操作zookeeper實現負載均衡

  導入所需要的依賴：

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.14</version>
        </dependency>
        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.10</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

   創建socket服務器端模擬生產者，socket客戶端作為消費者，客戶端從zookeeper節點信息里面獲取所有節點的ip和端口，來確定去連接哪一個socket服務器

  創建socket服務端:

package com.izk;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

import org.I0Itec.zkclient.ZkClient;

//##ServerScoekt服務端
public class ZkServerScoekt implements Runnable {
    private int port = 8081;

    public static void main(String[] args) throws IOException {
        int port = 8081;
        ZkServerScoekt server = new ZkServerScoekt(port);
        Thread thread = new Thread(server);
        thread.start();
    }

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

    // 將服務信息注冊到注冊中心上去
    public void regServer() {
        ZkClient zkClient = new ZkClient("127.0.0.1:2181", 6000, 1000);
        String path = "/member/server-" + port;
        //檢測是否存在這個節點
        if (zkClient.exists(path)) {
            zkClient.delete(path);
        }
        String value="127.0.0.1:" + port;
        zkClient.createEphemeral(path, "127.0.0.1:" + port);
        System.out.println("##注冊節點信息值###"+value);
    }

    public void run() {
        ServerSocket serverSocket = null;
        try {
            serverSocket = new ServerSocket(port);
            regServer();
            System.out.println("Server start port:" + port);
            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) {

            }
        }
    }

}

 

 創建socket的客戶端(消費者)：

package com.izk;

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 {
    // 從zookeeper中獲取到的子節點的信息
    public static List<String> listServer = new ArrayList<String>();

    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();
        // 從zk獲取最新獲取的注冊服務連接
        final String memberServerPath = "/member";
        final ZkClient zkClient = new ZkClient("127.0.0.1:2181", 6000, 1000);
        // 獲當前下子節點
        List<String> children = zkClient.getChildren(memberServerPath);
        listServer.clear();
        for (String p : children) {
            // 讀取子節點value值
            listServer.add((String) zkClient.readData(memberServerPath + "/" + p));
        }
        System.out.println("最新服務信息listServer:" + listServer.toString());
        // 訂閱子節點事件
        zkClient.subscribeChildChanges(memberServerPath, new IZkChildListener() {
            // 子節點發生變化
            public void handleChildChange(String parentPath, List<String> childrens) throws Exception {
                listServer.clear();
                for (String subP : childrens) {
                    // 讀取子節點value值
                    listServer.add((String) zkClient.readData(memberServerPath + "/" + subP));
                }
            }
        });
    }

    // 服務調用次數
    private static int count = 1;
    // 會員服務集群數量
    private static int memberServerCount = 2;

    // 獲取當前server信息
    public static String getServer() {
        //輪訓采用的是取模算法
        String serverName = listServer.get(count % memberServerCount);
        ++count;
        return serverName;
    }

    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();
                }
            }
        }
    }
}

 

 

package com.izk;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

//ServerHandler
public class ServerHandler implements Runnable {
    private Socket socket;

    public ServerHandler(Socket socket) {
        this.socket = socket;
    }

    public void run() {
        BufferedReader in = null;
        PrintWriter out = null;
        try {
            in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
            out = new PrintWriter(this.socket.getOutputStream(), true);
            String body = null;
            while (true) {
                body = in.readLine();
                if (body == null)
                    break;
                System.out.println("Receive : " + body);
                out.println("Hello, " + body);
            }

        } catch (Exception e) {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
            if (out != null) {
                out.close();
            }
            if (this.socket != null) {
                try {
                    this.socket.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
                this.socket = null;
            }
        }
    }
}