怎么通过Java实现反向代理集群服务的平滑分配
1.理解全过程1.1.概述需要具备的知识Socket网络编程反向代理的理解平滑加权轮询算法的理解线程池的理解目的：实现Socket集群服务的平滑加权轮询负载。业务实现：客户端通过用户名来查询集群服务中的用户信息。1.2.整个流程客户端发起Socket请求给反向代理的Socket服务（客户端并不知道服务端是反向代理服务器）反向代理服务器接收到Socket服务请求线程池开启服务线程去处理请求线程服务通过平滑加权轮询算法寻找当前权重最高的下游服务通过负载均衡算法返回的服务节点信息来创建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 < 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 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;i2.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;i2.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;i3.开始测试
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次请求（测试提权）
先把宕机的服务启动起来，然后多测试几次，看看测试结果。可以看到，权重降低后又提起来了，说明测试提权成功。
还有两个点没测：第一个是一直降权后，会不会出现宕机的服务不再分配到？这就起到”剔除“宕机服务的效果？第二个是服务恢复后，会不会出现宕机再起的服务需要慢慢恢复权重，直到一定值后才可以分配到？
Shapr3D是一款面向工业设计、机械工程、建筑概念和三维打印工作流的CAD软件。Mac版采用Parasolid建模内核，支持草图约束、实体建模、工程图、可视化渲染及常见CAD格式交换，并可通过账户在多台设备之间同步项目。
REAPER是Cockos开发的数字音频工作站，提供多轨音频与MIDI录制、剪辑、处理、混音和母带制作工具。Mac版兼容Intel与Apple芯片，支持AU、VST、VST3、CLAP等插件格式，并提供高度可定制的工作流程。
Ableton Live 是面向音乐制作人与现场表演者的数字音频工作站，提供编曲视图、独具特色的现场视图、音频录制、MIDI创作、实时变速、乐器及效果器。Mac版原生支持Apple芯片，并可连接音频接口、MIDI控制器和第三方插件。
Adobe After Effects 是面向动态图形、影视特效与视频合成的专业创作软件。它提供关键帧动画、遮罩抠像、运动跟踪、三维空间、文字动画和表达式等工具，并可与 Premiere Pro、Photoshop、Illustrator
Adobe Audition 是面向播客、影视后期、音乐制作及内容创作者的专业音频工作站，提供波形精修、多轨录制与混音、频谱编辑、降噪修复、响度匹配和效果处理工具，并可与 Adobe Premiere 协同完成视频声音制作。
Adobe Bridge 是面向摄影师、设计师及内容团队的免费数字资产管理工具，可集中预览照片、视频和 Adobe 项目文件，并利用评级、标签、关键词、元数据、筛选与收藏集完成分类检索。它还支持批量重命名、格式导出、相机导入及 Camera
Adobe Illustrator 是 Adobe 推出的专业矢量图形设计软件，适合在 Mac 上制作标志、图标、插画、包装、信息图及印刷版面。它以路径和锚点构建可无损缩放的图形，并提供钢笔、形状、文字、图像描摹、渐变、画板及生成式功能。
Adobe InDesign 是面向印刷品与数字出版物的专业版面设计软件，可在 Mac 上制作书籍、杂志、宣传册、海报、报告及交互式文档。它提供精细的文字样式、网格、母版、长文档管理和印前输出能力。
Adobe Lightroom Classic 是面向桌面摄影工作流的照片管理与后期处理软件，可将原始照片保存在本地硬盘，通过目录、关键词、评分和收藏夹高效整理图库，并提供 RAW 冲印、蒙版、镜头校正、批量同步及多格式导出等功能。
Adobe Media Encoder 是 Adobe 推出的专业音视频编码与转码工具，可通过队列、预设和监视文件夹批量完成格式转换、代理文件创建及多平台成片输出，并与 Premiere Pro、After Effects 等软件紧密协作。
Adobe Photoshop 是面向摄影师、设计师和内容创作者的专业图像处理软件，提供图层、蒙版、选区、修复、调色、文字排版、智能对象及生成式编辑等能力。Mac 版支持 Intel 与 Apple Silicon 处理器。
Adobe Premiere Pro 是面向专业视频制作的非线性剪辑软件，提供多轨时间线、调色、音频处理、字幕、代理工作流和多格式输出能力，并可与 After Effects、Audition、Photoshop 及 Frame.io 协同
Photoshop 2026 是 Adobe 推出的专业图像处理与视觉设计软件，支持 Windows、macOS 和 iPad 等平台，广泛应用于摄影修图、电商设计、平面海报、数字绘画及视觉合成等创作场景。
Blender 是一款免费开源、跨平台的专业 3D 创作软件，集建模、动画、渲染、视频编辑与视觉合成等功能于一体，广泛应用于影视动画、游戏设计和建筑可视化等领域。软件支持 Cycles 物理渲染器与 Eevee 实时渲染引擎，并提供多边形建模、骨骼绑定、物理模拟等专业工具。Blender 兼容 Windows、macOS 和 Linux 系统，安装包轻巧、运行流畅，依托活跃的全球开发者社区持续更新，是从初学者到专业创作者都值得选择的正版 3D 创作工具。
极度公式是一款跨平台专业LaTeX公式识别编辑软件，支持OCR公式识别和多平台编辑。和使用说明，避免使用，享受完整功能与稳定支持。做扫描整理、文字提取和表格转换时，它能把识别后的处理步骤接得更顺，资料录入这类场景会省下不少时间。
一款文字处理软件，一种订阅式的跨平台办公软件，基于云平台提供多种服务，通过将 Excel 和 Outlook 等应用与 OneDrive 和 Microsoft Teams 等强大的云服务相结合，Office 365 可让任何人使用任何设备随时随地创建和共享内容。
Wise Folder Hider Pro 是一款专业级文件和文件夹隐藏加密软件，为私密数据添加多重保护。高频操作更强调就近处理，浏览、整理和跨目录移动文件时，来回切换和重复点击都会少很多。
CodeExpander 是一款快捷短语输入增强工具，通过键入缩写自动展开为自定义文段，提升工作效率。任务管理和过程控制会更完整，持续下载、批量同步或需要稳定传输流程的场景会更适合它。