通過Java實(shí)現(xiàn)反向代理集群服務(wù)的平滑分配
1.理解全過程
1.1.概述
需要具備的知識
- Socket網(wǎng)絡(luò)編程
- 反向代理的理解
- 平滑加權(quán)輪詢算法的理解
- 線程池的理解
目的:實(shí)現(xiàn)Socket 集群服務(wù)的平滑加權(quán)輪詢負(fù)載。
業(yè)務(wù)實(shí)現(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ù),對客戶端無感,客戶端并不知道具體訪問了哪個真實(shí)服務(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.代碼實(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,不超過配置的起始權(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("真實(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ù)類
通過線程池來管理代理服務(wù)線程,不同的請求分發(fā)到不同線程處理。這里用的是newCachedThreadPool 線程池。
代理服務(wù)器在本地啟動。這里也可以新建一個類來啟動服務(wù),這樣可以啟動多個代理服務(wù),這里為了簡單就直接在本類的main方法啟動。可以對比后面的業(yè)務(wù)服務(wù)類,業(yè)務(wù)服務(wù)類就是這么起的,因?yàn)橐诺讲煌姆?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ù)實(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ù)的類,通過用戶名稱簡單獲取信息
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)有問題?
其實(shí)不是,是打印信息的位置沒放對……。要在下次訪問才可以看到上一次的降權(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ò)編程等,實(shí)現(xiàn)了反向代理集群服務(wù)的平滑分配,并通過降權(quán)/提權(quán)實(shí)現(xiàn)宕機(jī)服務(wù)的”剔除“和緩沖恢復(fù)。
到此這篇關(guān)于通過Java實(shí)現(xiàn)反向代理集群服務(wù)的平滑分配的文章就介紹到這了,更多相關(guān)Java集群服務(wù)平滑分配內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
SpringSecurity6自定義JSON登錄的實(shí)現(xiàn)
目前最新版的Spring Boot已經(jīng)到了3.0.5了,隨之而來Spring Security 目前的版本也到了6.0.2了,Spring Security寫法的變化特別多,本文就來介紹下,感興趣的可以了解一下2023-12-12Springboot如何統(tǒng)一處理Filter異常
這篇文章主要介紹了Springboot如何統(tǒng)一處理Filter異常問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12Mybatis CURD及模糊查詢功能的實(shí)現(xiàn)
這篇文章主要介紹了Mybatis CURD及模糊查詢功能的實(shí)現(xiàn),有查詢刪除,插入,更新功能,通過實(shí)例代碼講解的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2021-06-06SpringBoot根據(jù)目錄結(jié)構(gòu)自動生成路由前綴的實(shí)現(xiàn)代碼
本文介紹如何根據(jù)目錄結(jié)構(gòu)給RequestMapping添加路由前綴,具體實(shí)現(xiàn)方法,本文通過示例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友參考下吧2021-08-08Java基于Spire Cloud Excel把Excel轉(zhuǎn)換成PDF
這篇文章主要介紹了Java基于Spire Cloud Excel把Excel轉(zhuǎn)換成PDF,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-05-05