天天看點
傳回

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

作者：程式那點事

學會了負載均衡算法，卻沒有用起來？

今天就來實戰一遍，感受下平滑權重輪詢算法的魅力。

通過Java語言，自己編寫的平滑權重輪詢算法，結合線程池和Socket 網絡程式設計等，實作了反向代理叢集服務的平滑配置設定，并通過降級/提權實作當機服務的”剔除“和緩沖恢複。

1.了解全過程

1.1.概述

需要具備的知識

  • Socket網絡程式設計
  • 反向代理的了解
  • 平滑權重輪詢算法的了解
  • 線程池的了解

目的：實作Socket 叢集服務的平滑權重輪詢負載。

業務實作：用戶端通過使用者名來查詢叢集服務中的使用者資訊。

1.2.整個流程

  1. 用戶端發起Socket請求給反向代理的Socket服務（用戶端并不知道服務端是反向代理伺服器）
  2. 反向代理伺服器接收到Socket服務請求
  3. 線程池開啟服務線程去處理請求
  4. 線程服務通過平滑權重輪詢算法尋找目前權重最高的下遊服務
  5. 通過負載均衡算法傳回的服務節點資訊來建立Socket請求
  6. 反向代理伺服器使用用戶端資訊，發起Socket請求給下遊服務
  7. Socket叢集服務節點收到Socket請求，查詢使用者資訊，再将處理結果傳回給反向代理伺服器
  8. 反向代理伺服器再将結果傳回給用戶端。
自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

幾個細節點

  • 使用反向代理服務，對用戶端無感，用戶端并不知道具體通路了哪個真實伺服器；
  • 反向代理伺服器每次通路下遊服務失敗時，就會降低該下遊伺服器的有效權重；每次通路下遊服務成功時，就會提高該下遊伺服器的有效權重（不超過配置的權重值）；
  • 平滑權重輪詢算法會對當機服務降級和提權，起到”剔除“當機服務和緩沖恢複當機服務的效果；
  • 反向代理伺服器重新開機後，所有配置恢複為配置參數；
  • 反向代理伺服器使用線程池釋出Socket服務，支援多個用戶端同時請求同時分發。

2.代碼實作

2.1.節點類

用于儲存服務節點的相關資訊

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;
    // 預設權重為：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;
    }
    // 每成功一次，恢複有效權重1，不超過配置的起始權重
    public void onInvokeSuccess(){
        if(effectiveWeight < weight) effectiveWeight++;
    }
    // 每失敗一次，有效權重減少1，無底線的減少
    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的配置檔案資訊來完成配置。模拟為了簡單就直接寫代碼裡了。

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<>();
    // 在配置檔案中讀取：節點集合資訊。如果在同一台伺服器測試，那就将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.負載均衡算法接口 

package com.yty.proxy.lba;

public interface Robin {

    Node selectNode();
}

2.4.平滑權重輪詢算法

詳細介紹可以閱讀前兩篇負載均衡算法的文章

package com.yty.proxy.lba;

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

/**
 * 權重輪詢算法：平滑權重輪詢算法
 */
public class WeightedRoundRobin implements Robin {

    private static List<Node> nodes;
    // 讀取配置資訊
    static {
        nodes = ProxyConfig.getProxyConfig();
    }
    /**
     * 按照目前權重（currentWeight）最大值擷取IP
     * @return Node
     */
    public Node selectNode(){
        if (nodes ==null || nodes.size()<=0) return null;
        if (nodes.size() == 1)  return nodes.get(0);

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

        synchronized (nodes){
            // 選出目前權重最大的節點
            Node nodeOfMaxWeight = null;
            for (Node node : nodes) {
                if (nodeOfMaxWeight == null)
                    nodeOfMaxWeight = node;
                else
                    nodeOfMaxWeight = nodeOfMaxWeight.compareTo(node) > 0 ? nodeOfMaxWeight : node;
            }
            // 平滑負載均衡
            nodeOfMaxWeight.setCurrentWeight(nodeOfMaxWeight.getCurrentWeight() - totalWeight);
            nodes.forEach(node -> node.setCurrentWeight(node.getCurrentWeight()+node.getEffectiveWeight()));
            return nodeOfMaxWeight;
        }
    }

}

2.5.代理服務線程類

用于處理代理服務請求的線程類，不同請求建立不同的線程來處理

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 {
        // 代理接收用戶端請求
        byte[] proxyDataBytes =null;
        proxyDataBytes = getData(proxyIn);
        System.out.println("代理收到請求資料："+new String(proxyDataBytes));
        if (proxyDataBytes == null){
            proxyOut.write("請求内容異常".getBytes());
        }

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

        // 代理響應用戶端
        assert serverData != null;
        proxyOut.write(serverData);
        proxySocket.shutdownOutput();
        System.out.println("代理響應用戶端資料："+new String(proxyDataBytes));
    }

    private byte[] dispatcherService(byte[] proxyDataBytes){
        // 選擇節點：發送請求和接收響應資訊
        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("真實服務端響應資料："+ new String(serverData));
            node.onInvokeSuccess();//提權
        } catch (IOException e) {
            node.onInvokeFault();//降級
            serverData = "代理的下遊伺服器異常".getBytes();
        }
        System.out.println("負載均衡到：" + 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("代理服務關閉socket資源異常");
        }
    }
}

2.6.代理服務類

通過線程池來管理代理服務線程，不同的請求分發到不同的線程處理。這裡用的是newCachedThreadPool 線程池。

代理伺服器在本地啟動。這裡也可以建立一個類來啟動服務，這樣可以啟動多個代理服務，這裡為了簡單就直接在本類的main方法啟動。可以對比後面的業務服務類，業務服務類就是這麼起的，因為要放到不同的伺服器啟動業務服務。

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("開啟代理服務……");
        proxyServer.start();
    }
}

2.7.業務實體類 

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.業務類

處理具體業務的類，通過使用者名稱簡單擷取資訊

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.業務服務線程類

用于處理業務服務請求的線程類，不同請求建立不同線程來處理

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 {
        // 接收用戶端請求
        byte[] dataBytes =null;
        dataBytes = getData(serverIn);
        if (dataBytes == null){
            serverOut.write("請求内容異常".getBytes());
        }
        String username = new String(dataBytes);
        System.out.println("收到請求資料："+username);

        // 具體業務代碼
        MyUserService myUserService = new MyUserService();
        MyUser user = myUserService.findByUsername(new String(dataBytes));
        String serverData = "沒有查詢到使用者" + username + "的資料";
        if(user!=null){
            serverData = user.toString();
            serverOut.write(user.toString().getBytes());
        }
        System.out.println("響應用戶端資料：" + 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("服務關閉socket資源異常");
        }
    }
}

2.10.業務服務類

通過線程池來管理業務服務線程，不同的請求分發到不同線程處理。這裡用的也是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.啟動三個業務服務(服務叢集)

本次測試是分在兩台伺服器測試，1台【192.168.233.100】、另外都是本地【127.0.0.1】。如果覺得麻煩，那就都配成本地【127.0.0.1】，然後起服務都在本地起。

服務1：在IP為192.168.233.100 的伺服器啟動

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

服務2：在本地伺服器啟動

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

服務3：在本地伺服器啟動

package com.yty.proxy.test;
import com.yty.proxy.server.SocketServer;
public class StartServer3 {
    public static void main(String[] args) {
        System.out.println("開啟後端服務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();
        // 發送資料
        out.write("小白".getBytes());
        out.flush();
        socket.shutdownOutput();
        // 讀取資料
        byte[] data = new Client().getData(in);
        System.out.println("響應資料："+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.開始測試

3.1.啟動所有服務

在代理配置類（ProxyConfig）中指定的伺服器啟動三個業務服務；

在你喜歡的伺服器中啟動代理服務（ProxyServer），這裡在本地啟動【127.0.0.1】；

用戶端在本地測試咯（IP必須是代理伺服器的IP，這裡測試的代理伺服器IP是【127.0.0.1】。

所有服務啟動後的截圖：

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

3.2.用戶端發起第一次請求

正常命中權重最高的節點2服務：節點資訊在代理伺服器中列印出來了【127.0.0.1、8002】。這些日志資訊正常情況是寫入到日志檔案，這裡隻在控制台列印出來。

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

3.3.用戶端發起第二次請求

改了使用者名再請求試試，發現忘記列印請求資料了……

第二次命中節點3服務，跟平滑權重算法預定的結果一樣。

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

3.4.用戶端發起第三次請求

這次命中了節點2：192.168.233.100，8001的服務。到此可以看到平滑權重輪詢算法正常運作中。

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

3.5.用戶端發起第四次請求（測試降級）

通過平滑權重輪詢算法運算，我們知道這次肯定命中節點2服務。是以，在發起請求前，先關閉節點2服務，再由用戶端發起請求。

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

細心的應該發現，有效權重沒變小啊，是不是降級有問題？

其實不是，是列印資訊的位置沒放對……。要在下次通路才可以看到上一次的降級結果，額，有點呆（上面的代碼我已經改了）。

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

3.6.用戶端發起第N次請求（測試提權）

先把當機的服務啟動起來，然後多測試幾次，看看測試結果。可以看到，權重降低後又提起來了，說明測試提權成功。

自己編寫平滑權重輪詢算法，實作反向代理叢集服務的平滑配置設定

還有兩個點沒測：第一個是一直降級後，會不會出現當機的服務不再配置設定到？這就起到”剔除“當機服務的效果？第二個是服務恢複後，會不會出現當機再起的服務需要慢慢恢複權重，直到一定值後才可以配置設定到？

4.結論

使用自己編寫的平滑權重輪詢算法，結合線程池和Socket 網絡程式設計等，實作了反向代理叢集服務的平滑配置設定，并通過降級/提權實作當機服務的”剔除“和緩沖恢複。

來源：https://www.cnblogs.com/dennyLee2025/p/16147207.html

上一篇: MQ系列：消息中間件執行原理
下一篇: 葉羅麗：冰公主的誕生和水王子有關，水王子擁有主宰生死的能力