通过Java实现反向代理集群服务的平滑分配

这篇文章主要介绍了如何通过Java语言，自己编写的平滑加权轮询算法，结合线程池和Socket 网络编程等，并实现反向代理集群服务的平滑分配，需要的可以参考一下

1.理解全过程

1.1.概述

需要具备的知识

Socket网络编程
反向代理的理解
平滑加权轮询算法的理解
线程池的理解

目的：实现Socket 集群服务的平滑加权轮询负载。

业务实现：客户端通过用户名来查询集群服务中的用户信息。

1.2.整个流程

客户端发起Socket请求给反向代理的Socket服务（客户端并不知道服务端是反向代理服务器）
反向代理服务器接收到Socket服务请求
线程池开启服务线程去处理请求
线程服务通过平滑加权轮询算法寻找当前权重最高的下游服务
通过负载均衡算法返回的服务节点信息来创建Socket请求
反向代理服务器使用客户端信息，发起Socket请求给下游服务
Socket集群服务节点收到Socket请求，查询用户信息，再将处理结果返回给反向代理服务器
反向代理服务器再将结果返回给客户端。

几个细节点

使用反向代理服务，对客户端无感，客户端并不知道具体访问了哪个真实服务器；
反向代理服务器每次访问下游服务失败时，就会降低该下游服务器器的有效权重；每次访问下游服务成功时，就会提高该下游服务器的有效权重（不超过配置的权重值）；
平滑加权轮询算法会对宕机服务降权和提权，起到”剔除“宕机服务和缓冲恢复宕机服务的效果；
反向代理服务器重启后，所有配置恢复为配置参数；
反向代理服务器使用线程池发布Socket服务，支持多个客户端同时请求同时分发。

2.代码实现

2.1.节点类

用于保存服务节点相关信息

package com.yty.proxy.lba; public class Node implements Comparable{ 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  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 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 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 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 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

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 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 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

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 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