怎么通過(guò)Java實(shí)現(xiàn)反向代理集群服務(wù)的平滑分配

本文小編為大家詳細(xì)介紹“怎么通過(guò)Java實(shí)現(xiàn)反向代理集群服務(wù)的平滑分配”，內(nèi)容詳細(xì)，步驟清晰，細(xì)節(jié)處理妥當(dāng)，希望這篇“怎么通過(guò)Java實(shí)現(xiàn)反向代理集群服務(wù)的平滑分配”文章能幫助大家解決疑惑，下面跟著小編的思路慢慢深入，一起來(lái)學(xué)習(xí)新知識(shí)吧。

1.理解全過(guò)程

1.1.概述

需要具備的知識(shí)

• Socket網(wǎng)絡(luò)編程

• 反向代理的理解

• 平滑加權(quán)輪詢算法的理解

• 線程池的理解

目的：實(shí)現(xiàn)Socket 集群服務(wù)的平滑加權(quán)輪詢負(fù)載。

業(yè)務(wù)實(shí)現(xiàn)：客戶端通過(guò)用戶名來(lái)查詢集群服務(wù)中的用戶信息。

1.2.整個(gè)流程

• 客戶端發(fā)起Socket請(qǐng)求給反向代理的Socket服務(wù)（客戶端并不知道服務(wù)端是反向代理服務(wù)器）

• 反向代理服務(wù)器接收到Socket服務(wù)請(qǐng)求

• 線程池開(kāi)啟服務(wù)線程去處理請(qǐng)求

• 線程服務(wù)通過(guò)平滑加權(quán)輪詢算法尋找當(dāng)前權(quán)重最高的下游服務(wù)

• 通過(guò)負(fù)載均衡算法返回的服務(wù)節(jié)點(diǎn)信息來(lái)創(chuàng)建Socket請(qǐng)求

• 反向代理服務(wù)器使用客戶端信息，發(fā)起Socket請(qǐng)求給下游服務(wù)

• Socket集群服務(wù)節(jié)點(diǎn)收到Socket請(qǐng)求，查詢用戶信息，再將處理結(jié)果返回給反向代理服務(wù)器

• 反向代理服務(wù)器再將結(jié)果返回給客戶端。

幾個(gè)細(xì)節(jié)點(diǎn)

• 使用反向代理服務(wù)，對(duì)客戶端無(wú)感，客戶端并不知道具體訪問(wèn)了哪個(gè)真實(shí)服務(wù)器；

• 反向代理服務(wù)器每次訪問(wèn)下游服務(wù)失敗時(shí)，就會(huì)降低該下游服務(wù)器器的有效權(quán)重；每次訪問(wèn)下游服務(wù)成功時(shí)，就會(huì)提高該下游服務(wù)器的有效權(quán)重（不超過(guò)配置的權(quán)重值）；

• 平滑加權(quán)輪詢算法會(huì)對(duì)宕機(jī)服務(wù)降權(quán)和提權(quán)，起到”剔除“宕機(jī)服務(wù)和緩沖恢復(fù)宕機(jī)服務(wù)的效果；

• 反向代理服務(wù)器重啟后，所有配置恢復(fù)為配置參數(shù)；

• 反向代理服務(wù)器使用線程池發(fā)布Socket服務(wù)，支持多個(gè)客戶端同時(shí)請(qǐng)求同時(shí)分發(fā)。

2.代碼實(shí)現(xiàn)

2.1.節(jié)點(diǎn)類

用于保存服務(wù)節(jié)點(diǎn)相關(guān)信息

package com.yty.proxy.lba;


public class Node implements Comparable<Node>{
    private String ip;
    private Integer port;
    private final Integer weight;
    private Integer effectiveWeight;
    private Integer currentWeight;
    // 默認(rèn)權(quán)重為：1
    public Node(String ip,Integer port){
        this(ip,port,1);
    }

    public Node(String ip,Integer port, Integer weight){
        this.ip = ip;
        this.port = port;
        this.weight = weight;
        this.effectiveWeight = weight;
        this.currentWeight = weight;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public Integer getWeight() {
        return weight;
    }

    public Integer getEffectiveWeight() {
        return effectiveWeight;
    }

    public void setEffectiveWeight(Integer effectiveWeight) {
        this.effectiveWeight = effectiveWeight;
    }

    public Integer getCurrentWeight() {
        return currentWeight;
    }

    public void setCurrentWeight(Integer currentWeight) {
        this.currentWeight = currentWeight;
    }
    // 每成功一次，恢復(fù)有效權(quán)重1，不超過(guò)配置的起始權(quán)重
    public void onInvokeSuccess(){
        if(effectiveWeight < weight) effectiveWeight++;
    }
    // 每失敗一次，有效權(quán)重減少1，無(wú)底線的減少
    public void onInvokeFault(){
        effectiveWeight--;
    }

    @Override
    public int compareTo(Node node) {
        return currentWeight > node.currentWeight ? 1 : (currentWeight.equals(node.currentWeight) ? 0 : -1);
    }

    @Override
    public String toString() {
        return "Node{" +
                "ip='" + ip + '\'' +
                ", port=" + port +
                ", weight=" + weight +
                ", effectiveWeight=" + effectiveWeight +
                ", currentWeight=" + currentWeight +
                '}';
    }
}

2.2.代理配置類

一般都是放在配置文件中配置，然后讀取指定key的配置文件信息來(lái)完成配置。模擬為了簡(jiǎn)單就直接寫代碼里了。

package com.yty.proxy;

import com.yty.proxy.lba.Node;

import java.util.ArrayList;
import java.util.List;

public class ProxyConfig {

    private static List<Node> nodes = new ArrayList<>();
    // 在配置文件中讀?。汗?jié)點(diǎn)集合信息。如果在同一臺(tái)服務(wù)器測(cè)試，那就將ip配成一樣
    static {
        nodes.add(new Node("192.168.233.100",8001,2));
        nodes.add(new Node("127.0.0.1",8002,5));
        nodes.add(new Node("127.0.0.1",8003,3));
    }
    public static List<Node> getProxyConfig(){
        return nodes;
    }
}

2.3.負(fù)載均衡算法接口

package com.yty.proxy.lba;

public interface Robin {

    Node selectNode();
}

2.4.平滑加權(quán)輪詢算法

詳細(xì)介紹可以閱讀前兩篇負(fù)載均衡算法的文章

package com.yty.proxy.lba;

import com.yty.proxy.ProxyConfig;
import java.util.List;

/**
 * 加權(quán)輪詢算法：平滑加權(quán)輪詢算法
 */
public class WeightedRoundRobin implements Robin {

    private static List<Node> nodes;
    // 讀取配置信息
    static {
        nodes = ProxyConfig.getProxyConfig();
    }
    /**
     * 按照當(dāng)前權(quán)重（currentWeight）最大值獲取IP
     * @return Node
     */
    public Node selectNode(){
        if (nodes ==null || nodes.size()<=0) return null;
        if (nodes.size() == 1)  return nodes.get(0);

        // 權(quán)重之和
        Integer totalWeight = 0;
        for(Node node : nodes){
            totalWeight += node.getEffectiveWeight();
        }

        synchronized (nodes){
            // 選出當(dāng)前權(quán)重最大的節(jié)點(diǎn)
            Node nodeOfMaxWeight = null;
            for (Node node : nodes) {
                if (nodeOfMaxWeight == null)
                    nodeOfMaxWeight = node;
                else
                    nodeOfMaxWeight = nodeOfMaxWeight.compareTo(node) > 0 ? nodeOfMaxWeight : node;
            }
            // 平滑負(fù)載均衡
            nodeOfMaxWeight.setCurrentWeight(nodeOfMaxWeight.getCurrentWeight() - totalWeight);
            nodes.forEach(node -> node.setCurrentWeight(node.getCurrentWeight()+node.getEffectiveWeight()));
            return nodeOfMaxWeight;
        }
    }

}

2.5.代理服務(wù)線程類

用于處理代理服務(wù)請(qǐng)求的線程類，不同請(qǐng)求創(chuàng)建不同線程來(lái)處理

package com.yty.proxy;

import java.io.*;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

public class ProxyServerThread implements Runnable {
    private Socket proxySocket;
    private OutputStream proxyOut;
    private InputStream proxyIn;
    private Socket socket;
    private OutputStream serverOut;
    private InputStream serverIn;
    public ProxyServerThread(Socket proxySocket) throws IOException {
        this.proxySocket = proxySocket;
        this.proxySocket.setSoTimeout(6000);
        this.proxyOut = proxySocket.getOutputStream();
        this.proxyIn = proxySocket.getInputStream();
    }

    @Override
    public void run() {
        try {
            this.proxyService();
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            this.close();
        }
    }

    private void proxyService() throws IOException {
        // 代理接收客戶端請(qǐng)求
        byte[] proxyDataBytes =null;
        proxyDataBytes = getData(proxyIn);
        System.out.println("代理收到請(qǐng)求數(shù)據(jù)："+new String(proxyDataBytes));
        if (proxyDataBytes == null){
            proxyOut.write("請(qǐng)求內(nèi)容異常".getBytes());
        }

        byte[] serverData = this.dispatcherService(proxyDataBytes);

        // 代理響應(yīng)客戶端
        assert serverData != null;
        proxyOut.write(serverData);
        proxySocket.shutdownOutput();
        System.out.println("代理響應(yīng)客戶端數(shù)據(jù)："+new String(proxyDataBytes));
    }

    private byte[] dispatcherService(byte[] proxyDataBytes){
        // 選擇節(jié)點(diǎn)：發(fā)送請(qǐng)求和接收響應(yīng)信息
        Robin wrr = new WeightedRoundRobin();
        Node node = wrr.selectNode();
        byte[] serverData = null;
        try {
            this.socket = new Socket(node.getIp(), node.getPort());
            socket.setSoTimeout(6000);
            serverIn = socket.getInputStream();
            serverOut= socket.getOutputStream();
            serverOut.write(proxyDataBytes);
            socket.shutdownOutput();
            serverData = getData(serverIn);
            System.out.println("真實(shí)服務(wù)端響應(yīng)數(shù)據(jù)："+ new String(serverData));
            node.onInvokeSuccess();//提權(quán)
        } catch (IOException e) {
            node.onInvokeFault();//降權(quán)
            serverData = "代理的下游服務(wù)器異常".getBytes();
        }
        System.out.println("負(fù)載均衡到：" + node);
        return serverData;
    }

    private byte[] getData(InputStream in) throws IOException {
        List<Byte> byteList = new ArrayList<>();
        int temp = -1;
        while (true) {
            temp = in.read();
            if (temp != -1)
                byteList.add((byte) temp);
            else
                break;
        }
        byte[] bytes = new byte[byteList.size()];
        for (int i=0;i<byteList.size();i++){
            bytes[i]=byteList.get(i);
        }
        return bytes;
    }

    private void close() {
        try {
            if (proxySocket!=null){
                proxySocket.shutdownInput();
                proxySocket.close();
            }
            if (socket!=null){
                socket.shutdownInput();
                socket.close();
            }
        }catch (IOException e){
            e.printStackTrace();
            System.out.println("代理服務(wù)關(guān)閉socket資源異常");
        }
    }
}

2.6.代理服務(wù)類

通過(guò)線程池來(lái)管理代理服務(wù)線程，不同的請(qǐng)求分發(fā)到不同線程處理。這里用的是newCachedThreadPool 線程池。

代理服務(wù)器在本地啟動(dòng)。這里也可以新建一個(gè)類來(lái)啟動(dòng)服務(wù)，這樣可以啟動(dòng)多個(gè)代理服務(wù)，這里為了簡(jiǎn)單就直接在本類的main方法啟動(dòng)?？梢詫?duì)比后面的業(yè)務(wù)服務(wù)類，業(yè)務(wù)服務(wù)類就是這么起的，因?yàn)橐诺讲煌姆?wù)器啟動(dòng)業(yè)務(wù)服務(wù)。

package com.yty.proxy;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ProxyServer {
    private final Integer port;
    private ServerSocket serverSocket;
    public ProxyServer(Integer port) {
        this.port = port;
    }

    public void start(){
        ExecutorService threadPool = Executors.newCachedThreadPool();
        try {
            serverSocket = new ServerSocket(port);
            while (true){
                Socket socket = serverSocket.accept();
                threadPool.execute(new ProxyServerThread(socket));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    public static void main(String[] args) {
        Integer proxyPort=8000;
        ProxyServer proxyServer = new ProxyServer(proxyPort);
        System.out.println("開(kāi)啟代理服務(wù)……");
        proxyServer.start();
    }
}

2.7.業(yè)務(wù)實(shí)體類

package com.yty.proxy.server;

public class MyUser {
    private String name;
    private int age;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public MyUser(String name, int age) {
        this.name = name;
        this.age = age;
    }

    public MyUser() {
    }

    @Override
    public String toString() {
        return "{" +"name='" + name + '\'' + ", age=" + age + '}';
    }
}

2.8.業(yè)務(wù)類

處理具體業(yè)務(wù)的類，通過(guò)用戶名稱簡(jiǎn)單獲取信息

package com.yty.proxy.server;

import java.util.ArrayList;
import java.util.List;

public class MyUserService {
    private static List<MyUser> list = new ArrayList<>();

    static{
        list.add(new MyUser("張三",18));
        list.add(new MyUser("張三豐",38));
        list.add(new MyUser("小白",18));
    }
    public MyUser findByUsername(String username){
        for (MyUser user:list){
            if (user.getName().equals(username)){
                return user;
            }
        }
        return null;
    }
}

2.9.業(yè)務(wù)服務(wù)線程類

用于處理業(yè)務(wù)服務(wù)請(qǐng)求的線程類，不同請(qǐng)求創(chuàng)建不同線程來(lái)處理

package com.yty.proxy.server;

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

public class SocketServerThread implements Runnable {
    private Socket socket;
    private OutputStream serverOut;
    private InputStream serverIn;
    public SocketServerThread(Socket socket) throws IOException {
        this.socket = socket;
        socket.setSoTimeout(6000);
        this.serverOut = socket.getOutputStream();
        this.serverIn = socket.getInputStream();
    }

    @Override
    public void run() {
        try {
            this.service();
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            this.close();
        }
    }
    private void service() throws IOException {
        // 接收客戶端請(qǐng)求
        byte[] dataBytes =null;
        dataBytes = getData(serverIn);
        if (dataBytes == null){
            serverOut.write("請(qǐng)求內(nèi)容異常".getBytes());
        }
        String username = new String(dataBytes);
        System.out.println("收到請(qǐng)求數(shù)據(jù)："+username);

        // 具體業(yè)務(wù)代碼
        MyUserService myUserService = new MyUserService();
        MyUser user = myUserService.findByUsername(new String(dataBytes));
        String serverData = "沒(méi)有查詢到用戶" + username + "的數(shù)據(jù)";
        if(user!=null){
            serverData = user.toString();
            serverOut.write(user.toString().getBytes());
        }
        System.out.println("響應(yīng)客戶端數(shù)據(jù)：" + serverData);
    }

    private byte[] getData(InputStream in) throws IOException {
        List<Byte> byteList = new ArrayList<>();
        int temp = -1;
        while (true) {
            temp = in.read();
            if (temp != -1)
                byteList.add((byte) temp);
            else
                break;
        }
        byte[] bytes = new byte[byteList.size()];
        for (int i=0;i<byteList.size();i++){
            bytes[i]=byteList.get(i);
        }
        return bytes;
    }

    private void close() {
        try {
            if (socket!=null){
                socket.shutdownInput();
                socket.shutdownOutput();
                socket.close();
            }
        }catch (IOException e){
            e.printStackTrace();
            System.out.println("服務(wù)關(guān)閉socket資源異常");
        }
    }
}

2.10.業(yè)務(wù)服務(wù)類

通過(guò)線程池來(lái)管理業(yè)務(wù)服務(wù)線程，不同的請(qǐng)求分發(fā)到不同線程處理。這里用的也是newCachedThreadPool 線程池。

package com.yty.proxy.server;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SocketServer {
    private final Integer port;
    private ServerSocket serverSocket;
    private Integer threads = 3;


    public SocketServer(Integer port) {
        this.port = port;
    }

    public void start(){
        ExecutorService threadPool = Executors.newFixedThreadPool(threads);
        try {
            serverSocket = new ServerSocket(port);
            while (true){
                Socket socket = serverSocket.accept();
                threadPool.execute(new SocketServerThread(socket));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

}

2.11.啟動(dòng)三個(gè)業(yè)務(wù)服務(wù)(服務(wù)集群)

本次測(cè)試是分在兩臺(tái)服務(wù)器測(cè)試，1臺(tái)【192.168.233.100】、另外都是本地【127.0.0.1】。如果覺(jué)得麻煩，那就都配成本地【127.0.0.1】，然后起服務(wù)都在本地起。

服務(wù)1：在IP為192.168.233.100 的服務(wù)器啟動(dòng)

package com.yty.proxy.test;
import com.yty.proxy.server.SocketServer;
public class StartServer1 {
    public static void main(String[] args) {
        System.out.println("開(kāi)啟后端服務(wù)8001……");
        new SocketServer(8001).start();
    }
}

服務(wù)2：在本地服務(wù)器啟動(dòng)

package com.yty.proxy.test;
import com.yty.proxy.server.SocketServer;
public class StartServer2 {
    public static void main(String[] args) {
        System.out.println("開(kāi)啟后端服務(wù)8002……");
        new SocketServer(8002).start();
    }
}

服務(wù)3：在本地服務(wù)器啟動(dòng)

package com.yty.proxy.test;
import com.yty.proxy.server.SocketServer;
public class StartServer3 {
    public static void main(String[] args) {
        System.out.println("開(kāi)啟后端服務(wù)8003……");
        new SocketServer(8003).start();
    }
}

2.12.客戶端

package com.yty.proxy.test;

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

public class Client {
    public static void main(String[] args) throws IOException {
        String ip = "127.0.0.1";
        int port = 8000;
        Socket socket = new Socket(ip, port);
        socket.setSoTimeout(6000);
        OutputStream out = socket.getOutputStream();
        InputStream in = socket.getInputStream();
        // 發(fā)送數(shù)據(jù)
        out.write("小白".getBytes());
        out.flush();
        socket.shutdownOutput();
        // 讀取數(shù)據(jù)
        byte[] data = new Client().getData(in);
        System.out.println("響應(yīng)數(shù)據(jù)："+new String(data));
        out.close();
    }

    private byte[] getData(InputStream in) throws IOException {
        BufferedInputStream bin = new BufferedInputStream(in);
        List<Byte> byteList = new ArrayList<>();
        while (true) {
            int temp = bin.read();
            if (temp != -1)
                byteList.add((byte) temp);
            else
                break;
        }
        byte[] bytes = new byte[byteList.size()];
        for (int i=0;i<byteList.size();i++){
            bytes[i]=byteList.get(i);
        }
        return bytes;
    }
}

3.開(kāi)始測(cè)試

3.1.啟動(dòng)所有服務(wù)

在代理配置類（ProxyConfig）中指定的服務(wù)器啟動(dòng)三個(gè)業(yè)務(wù)服務(wù)；

在你喜歡的服務(wù)器中啟動(dòng)代理服務(wù)（ProxyServer），這里在本地啟動(dòng)【127.0.0.1】；

客戶端在本地測(cè)試咯（IP必須是代理服務(wù)器的IP，這里測(cè)試的代理服務(wù)器IP是【127.0.0.1】。

所有服務(wù)啟動(dòng)后的截圖：

3.2.客戶端發(fā)起第一次請(qǐng)求

正常命中權(quán)重最高的節(jié)點(diǎn)2服務(wù)：節(jié)點(diǎn)信息在代理服務(wù)器中打印出來(lái)了【127.0.0.1、8002】。這些日志信息正常情況是寫入到日志文件，這里只在控制臺(tái)打印出來(lái)。

3.3.客戶端發(fā)起第二次請(qǐng)求

改了用戶名再請(qǐng)求試試，發(fā)現(xiàn)忘記打印請(qǐng)求數(shù)據(jù)了……

第二次命中節(jié)點(diǎn)3服務(wù)，跟平滑加權(quán)算法預(yù)定的結(jié)果一樣。

3.4.客戶端發(fā)起第三次請(qǐng)求

這次命中了節(jié)點(diǎn)2：192.168.233.100，8001的服務(wù)。到此可以看到平滑加權(quán)輪詢算法正常運(yùn)作中。

3.5.客戶端發(fā)起第四次請(qǐng)求（測(cè)試降權(quán)）

通過(guò)平滑加權(quán)輪詢算法運(yùn)算，我們知道這次肯定命中節(jié)點(diǎn)2服務(wù)。所以，在發(fā)起請(qǐng)求前，先關(guān)閉節(jié)點(diǎn)2服務(wù)，再由客戶端發(fā)起請(qǐng)求。

細(xì)心的應(yīng)該發(fā)現(xiàn)，有效權(quán)重沒(méi)變小啊，是不是降權(quán)有問(wèn)題？

其實(shí)不是，是打印信息的位置沒(méi)放對(duì)……。要在下次訪問(wèn)才可以看到上一次的降權(quán)結(jié)果，額，有點(diǎn)呆（上面的代碼我已經(jīng)改了）。

3.6.客戶端發(fā)起第N次請(qǐng)求（測(cè)試提權(quán)）

先把宕機(jī)的服務(wù)啟動(dòng)起來(lái)，然后多測(cè)試幾次，看看測(cè)試結(jié)果。可以看到，權(quán)重降低后又提起來(lái)了，說(shuō)明測(cè)試提權(quán)成功。

還有兩個(gè)點(diǎn)沒(méi)測(cè)：第一個(gè)是一直降權(quán)后，會(huì)不會(huì)出現(xiàn)宕機(jī)的服務(wù)不再分配到？這就起到”剔除“宕機(jī)服務(wù)的效果？第二個(gè)是服務(wù)恢復(fù)后，會(huì)不會(huì)出現(xiàn)宕機(jī)再起的服務(wù)需要慢慢恢復(fù)權(quán)重，直到一定值后才可以分配到？

讀到這里，這篇“怎么通過(guò)Java實(shí)現(xiàn)反向代理集群服務(wù)的平滑分配”文章已經(jīng)介紹完畢，想要掌握這篇文章的知識(shí)點(diǎn)還需要大家自己動(dòng)手實(shí)踐使用過(guò)才能領(lǐng)會(huì)，如果想了解更多相關(guān)內(nèi)容的文章，歡迎關(guān)注億速云行業(yè)資訊頻道。