通過Java實現(xiàn)反向代理集群服務(wù)的平滑分配
更新時間：2022年04月15日 08:08:26 作者：淵渟岳
這篇文章主要介紹了如何通過Java語言，自己編寫的平滑加權(quán)輪詢算法，結(jié)合線程池和Socket?網(wǎng)絡(luò)編程等，并實現(xiàn)反向代理集群服務(wù)的平滑分配，需要的可以參考一下
1.理解全過程
1.1.概述
需要具備的知識
- Socket網(wǎng)絡(luò)編程
- 反向代理的理解
- 平滑加權(quán)輪詢算法的理解
- 線程池的理解
目的：實現(xiàn)Socket 集群服務(wù)的平滑加權(quán)輪詢負(fù)載。
業(yè)務(wù)實現(xiàn)：客戶端通過用戶名來查詢集群服務(wù)中的用戶信息。
1.2.整個流程
- 客戶端發(fā)起Socket請求給反向代理的Socket服務(wù)（客戶端并不知道服務(wù)端是反向代理服務(wù)器）
- 反向代理服務(wù)器接收到Socket服務(wù)請求
- 線程池開啟服務(wù)線程去處理請求
- 線程服務(wù)通過平滑加權(quán)輪詢算法尋找當(dāng)前權(quán)重最高的下游服務(wù)
- 通過負(fù)載均衡算法返回的服務(wù)節(jié)點(diǎn)信息來創(chuàng)建Socket請求
- 反向代理服務(wù)器使用客戶端信息，發(fā)起Socket請求給下游服務(wù)
- Socket集群服務(wù)節(jié)點(diǎn)收到Socket請求，查詢用戶信息，再將處理結(jié)果返回給反向代理服務(wù)器
- 反向代理服務(wù)器再將結(jié)果返回給客戶端。
幾個細(xì)節(jié)點(diǎn)
- 使用反向代理服務(wù)，對客戶端無感，客戶端并不知道具體訪問了哪個真實服務(wù)器；
- 反向代理服務(wù)器每次訪問下游服務(wù)失敗時，就會降低該下游服務(wù)器器的有效權(quán)重；每次訪問下游服務(wù)成功時，就會提高該下游服務(wù)器的有效權(quán)重（不超過配置的權(quán)重值）；
- 平滑加權(quán)輪詢算法會對宕機(jī)服務(wù)降權(quán)和提權(quán)，起到”剔除“宕機(jī)服務(wù)和緩沖恢復(fù)宕機(jī)服務(wù)的效果；
- 反向代理服務(wù)器重啟后，所有配置恢復(fù)為配置參數(shù)；
- 反向代理服務(wù)器使用線程池發(fā)布Socket服務(wù)，支持多個客戶端同時請求同時分發(fā)。
2.代碼實現(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，不超過配置的起始權(quán)重
public void onInvokeSuccess(){
if(effectiveWeight < weight) effectiveWeight++;
}
// 每失敗一次，有效權(quán)重減少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<>();
// 在配置文件中讀?。汗?jié)點(diǎn)集合信息。如果在同一臺服務(wù)器測試，那就將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ù)請求的線程類，不同請求創(chuàng)建不同線程來處理
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("代理收到請求數(shù)據(jù)："+new String(proxyDataBytes));
if (proxyDataBytes == null){
proxyOut.write("請求內(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ā)送請求和接收響應(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("真實服務(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ù)類
通過線程池來管理代理服務(wù)線程，不同的請求分發(fā)到不同線程處理。這里用的是newCachedThreadPool 線程池。
代理服務(wù)器在本地啟動。這里也可以新建一個類來啟動服務(wù)，這樣可以啟動多個代理服務(wù)，這里為了簡單就直接在本類的main方法啟動?？梢詫Ρ群竺娴?strong>業(yè)務(wù)服務(wù)類，業(yè)務(wù)服務(wù)類就是這么起的，因為要放到不同的服務(wù)器啟動業(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("開啟代理服務(wù)……");
proxyServer.start();
}
}
2.7.業(yè)務(wù)實體類
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ù)的類，通過用戶名稱簡單獲取信息
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ù)請求的線程類，不同請求創(chuàng)建不同線程來處理
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("請求內(nèi)容異常".getBytes());
}
String username = new String(dataBytes);
System.out.println("收到請求數(shù)據(jù)："+username);
// 具體業(yè)務(wù)代碼
MyUserService myUserService = new MyUserService();
MyUser user = myUserService.findByUsername(new String(dataBytes));
String serverData = "沒有查詢到用戶" + 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ù)類
通過線程池來管理業(yè)務(wù)服務(wù)線程，不同的請求分發(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.啟動三個業(yè)務(wù)服務(wù)(服務(wù)集群)
本次測試是分在兩臺服務(wù)器測試，1臺【192.168.233.100】、另外都是本地【127.0.0.1】。如果覺得麻煩，那就都配成本地【127.0.0.1】，然后起服務(wù)都在本地起。
服務(wù)1：在IP為192.168.233.100 的服務(wù)器啟動
package com.yty.proxy.test;
import com.yty.proxy.server.SocketServer;
public class StartServer1 {
public static void main(String[] args) {
System.out.println("開啟后端服務(wù)8001……");
new SocketServer(8001).start();
}
}
服務(wù)2：在本地服務(wù)器啟動
package com.yty.proxy.test;
import com.yty.proxy.server.SocketServer;
public class StartServer2 {
public static void main(String[] args) {
System.out.println("開啟后端服務(wù)8002……");
new SocketServer(8002).start();
}
}
服務(wù)3：在本地服務(wù)器啟動
package com.yty.proxy.test;
import com.yty.proxy.server.SocketServer;
public class StartServer3 {
public static void main(String[] args) {
System.out.println("開啟后端服務(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.開始測試
3.1.啟動所有服務(wù)
在代理配置類（ProxyConfig）中指定的服務(wù)器啟動三個業(yè)務(wù)服務(wù)；
在你喜歡的服務(wù)器中啟動代理服務(wù)（ProxyServer），這里在本地啟動【127.0.0.1】；
客戶端在本地測試咯（IP必須是代理服務(wù)器的IP，這里測試的代理服務(wù)器IP是【127.0.0.1】。
所有服務(wù)啟動后的截圖：
3.2.客戶端發(fā)起第一次請求
正常命中權(quán)重最高的節(jié)點(diǎn)2服務(wù)：節(jié)點(diǎn)信息在代理服務(wù)器中打印出來了【127.0.0.1、8002】。這些日志信息正常情況是寫入到日志文件，這里只在控制臺打印出來。
3.3.客戶端發(fā)起第二次請求
改了用戶名再請求試試，發(fā)現(xiàn)忘記打印請求數(shù)據(jù)了……
第二次命中節(jié)點(diǎn)3服務(wù)，跟平滑加權(quán)算法預(yù)定的結(jié)果一樣。
3.4.客戶端發(fā)起第三次請求
這次命中了節(jié)點(diǎn)2：192.168.233.100，8001的服務(wù)。到此可以看到平滑加權(quán)輪詢算法正常運(yùn)作中。
3.5.客戶端發(fā)起第四次請求（測試降權(quán)）
通過平滑加權(quán)輪詢算法運(yùn)算，我們知道這次肯定命中節(jié)點(diǎn)2服務(wù)。所以，在發(fā)起請求前，先關(guān)閉節(jié)點(diǎn)2服務(wù)，再由客戶端發(fā)起請求。
細(xì)心的應(yīng)該發(fā)現(xiàn)，有效權(quán)重沒變小啊，是不是降權(quán)有問題？
其實不是，是打印信息的位置沒放對……。要在下次訪問才可以看到上一次的降權(quán)結(jié)果，額，有點(diǎn)呆（上面的代碼我已經(jīng)改了）。
3.6.客戶端發(fā)起第N次請求（測試提權(quán)）
先把宕機(jī)的服務(wù)啟動起來，然后多測試幾次，看看測試結(jié)果?？梢钥吹?，權(quán)重降低后又提起來了，說明測試提權(quán)成功。
還有兩個點(diǎn)沒測：第一個是一直降權(quán)后，會不會出現(xiàn)宕機(jī)的服務(wù)不再分配到？這就起到”剔除“宕機(jī)服務(wù)的效果？第二個是服務(wù)恢復(fù)后，會不會出現(xiàn)宕機(jī)再起的服務(wù)需要慢慢恢復(fù)權(quán)重，直到一定值后才可以分配到？
4.結(jié)論
使用自己編寫的平滑加權(quán)輪詢算法，結(jié)合線程池和Socket 網(wǎng)絡(luò)編程等，實現(xiàn)了反向代理集群服務(wù)的平滑分配，并通過降權(quán)/提權(quán)實現(xiàn)宕機(jī)服務(wù)的”剔除“和緩沖恢復(fù)。
到此這篇關(guān)于通過Java實現(xiàn)反向代理集群服務(wù)的平滑分配的文章就介紹到這了,更多相關(guān)Java集群服務(wù)平滑分配內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
之前我們使用spring,最多的還是通過xml配置文件的方式來配置spring bean等內(nèi)容,隨著注解的廣泛應(yīng)用和spring4中Java config的引入,xml配置文件方式逐步被替換,但是如果是想要使用xml配置文件方式的話,也可以通過@ImportResource注解來實現(xiàn),下面我們來一起看下如何使用.2021-05-05
這篇文章主要介紹了Spring中的retry重試組件詳解,Retry重試組件是一個處理重試邏輯的工具,可以在出現(xiàn)異?；蚴∏闆r下自動進(jìn)行重試操作,從而提高程序的穩(wěn)定性和可靠性,需要的朋友可以參考下2023-10-10
這篇文章主要介紹了Java中的注解、元注解詳細(xì)解析,注解也叫元數(shù)據(jù),與類、接口、枚舉是在同一個層次,它可以聲明在包、類、字段、方法、局部變量、方法參數(shù)等的前面,用來對這些元素進(jìn)行說明,注釋,需要的朋友可以參考下2023-11-11
這篇文章主要介紹了Java原子變量類原理及實例解析,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2019-12-12
MyBatis 是支持定制化 SQL、存儲過程以及高級映射的優(yōu)秀的持久層框架。這篇文章主要介紹了MyBatis配置文件的寫法和簡單使用，需要的朋友參考下2017-01-01
Java中如何使用Redis GEO測算經(jīng)緯度距離
GEO也是Redis中提供的一種數(shù)據(jù)類型,可以將其理解成ZSet有序集合類型,本文給大家介紹Java-使用Redis GEO測算經(jīng)緯度距離,感興趣的朋友一起看看吧2023-11-11
springboot項目test文件夾下帶main方法的類不能運(yùn)行問題
這篇文章主要介紹了springboot項目test文件夾下帶main方法的類不能運(yùn)行問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-11-11
這篇文章主要為大家詳細(xì)介紹了java科研信息管理系統(tǒng)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-02-02
今天要跟大家聊一聊Java多線程的原理和使用方式,Java多線程是提高程序性能的重要機(jī)制,了解線程生命周期、同步機(jī)制、線程池和線程間通信等概念,可以幫助我們編寫出高效、可靠的多線程程序,需要的朋友可以參考下2024-07-07
springboot?通過博途獲取plc點(diǎn)位的數(shù)據(jù)代碼實現(xiàn)
這篇文章主要介紹了springboot?通過博途獲取plc點(diǎn)位的數(shù)據(jù)的代碼實現(xiàn),本文通過實例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2023-08-08