From 0b5ff2465b17b8d24410157e7b96311f4c996e14 Mon Sep 17 00:00:00 2001 From: Administrator Date: Fri, 25 Nov 2022 21:06:10 +0800 Subject: [PATCH] KLALB --- .classpath | 10 + .gitignore | 1 + .project | 17 ++ .settings/org.eclipse.core.resources.prefs | 2 + .settings/org.eclipse.jdt.core.prefs | 15 ++ src/org/kne/cloud/network/klalb/Consts.java | 5 + .../kne/cloud/network/klalb/KLALBBlock.java | 57 +++++ .../kne/cloud/network/klalb/KLALBClient.java | 66 ++++++ .../network/klalb/KLALBClientProtocol.java | 174 ++++++++++++++ .../kne/cloud/network/klalb/KLALBCore.java | 214 ++++++++++++++++++ .../kne/cloud/network/klalb/KLALBServer.java | 57 +++++ .../network/klalb/KLALBServerProtocol.java | 129 +++++++++++ .../cloud/network/klalb/TCPConnection.java | 72 ++++++ .../kne/cloud/network/klalb/TCPListener.java | 65 ++++++ .../kne/cloud/network/klalb/ThreadTool.java | 20 ++ src/org/kne/cloud/network/klalb/Tunnel.java | 52 +++++ 16 files changed, 956 insertions(+) create mode 100644 .classpath create mode 100644 .gitignore create mode 100644 .project create mode 100644 .settings/org.eclipse.core.resources.prefs create mode 100644 .settings/org.eclipse.jdt.core.prefs create mode 100644 src/org/kne/cloud/network/klalb/Consts.java create mode 100644 src/org/kne/cloud/network/klalb/KLALBBlock.java create mode 100644 src/org/kne/cloud/network/klalb/KLALBClient.java create mode 100644 src/org/kne/cloud/network/klalb/KLALBClientProtocol.java create mode 100644 src/org/kne/cloud/network/klalb/KLALBCore.java create mode 100644 src/org/kne/cloud/network/klalb/KLALBServer.java create mode 100644 src/org/kne/cloud/network/klalb/KLALBServerProtocol.java create mode 100644 src/org/kne/cloud/network/klalb/TCPConnection.java create mode 100644 src/org/kne/cloud/network/klalb/TCPListener.java create mode 100644 src/org/kne/cloud/network/klalb/ThreadTool.java create mode 100644 src/org/kne/cloud/network/klalb/Tunnel.java diff --git a/.classpath b/.classpath new file mode 100644 index 0000000..ae62d39 --- /dev/null +++ b/.classpath @@ -0,0 +1,10 @@ + + + + + + + + + + diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ae3c172 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/bin/ diff --git a/.project b/.project new file mode 100644 index 0000000..b5f524a --- /dev/null +++ b/.project @@ -0,0 +1,17 @@ + + + KLALB + + + + + + org.eclipse.jdt.core.javabuilder + + + + + + org.eclipse.jdt.core.javanature + + diff --git a/.settings/org.eclipse.core.resources.prefs b/.settings/org.eclipse.core.resources.prefs new file mode 100644 index 0000000..3b55ea5 --- /dev/null +++ b/.settings/org.eclipse.core.resources.prefs @@ -0,0 +1,2 @@ +eclipse.preferences.version=1 +encoding/=GBK diff --git a/.settings/org.eclipse.jdt.core.prefs b/.settings/org.eclipse.jdt.core.prefs new file mode 100644 index 0000000..17d114f --- /dev/null +++ b/.settings/org.eclipse.jdt.core.prefs @@ -0,0 +1,15 @@ +eclipse.preferences.version=1 +org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled +org.eclipse.jdt.core.compiler.codegen.methodParameters=do not generate +org.eclipse.jdt.core.compiler.codegen.targetPlatform=18 +org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve +org.eclipse.jdt.core.compiler.compliance=18 +org.eclipse.jdt.core.compiler.debug.lineNumber=generate +org.eclipse.jdt.core.compiler.debug.localVariable=generate +org.eclipse.jdt.core.compiler.debug.sourceFile=generate +org.eclipse.jdt.core.compiler.problem.assertIdentifier=error +org.eclipse.jdt.core.compiler.problem.enablePreviewFeatures=disabled +org.eclipse.jdt.core.compiler.problem.enumIdentifier=error +org.eclipse.jdt.core.compiler.problem.reportPreviewFeatures=warning +org.eclipse.jdt.core.compiler.release=enabled +org.eclipse.jdt.core.compiler.source=18 diff --git a/src/org/kne/cloud/network/klalb/Consts.java b/src/org/kne/cloud/network/klalb/Consts.java new file mode 100644 index 0000000..6b50a4f --- /dev/null +++ b/src/org/kne/cloud/network/klalb/Consts.java @@ -0,0 +1,5 @@ +package org.kne.cloud.network.klalb; + +public class Consts { + public static int BLOCKSIZE=65536; +} diff --git a/src/org/kne/cloud/network/klalb/KLALBBlock.java b/src/org/kne/cloud/network/klalb/KLALBBlock.java new file mode 100644 index 0000000..522939e --- /dev/null +++ b/src/org/kne/cloud/network/klalb/KLALBBlock.java @@ -0,0 +1,57 @@ +package org.kne.cloud.network.klalb; + +import java.util.Arrays; +import java.util.Objects; + +public class KLALBBlock implements Comparable{ + volatile byte[]data; + volatile int size; + volatile long number; + + + volatile long time=-1; + volatile Thread thread; + public KLALBBlock(byte[] b, int size,long number) { + data=b; + this.size=size; + this.number=number; + } + public KLALBBlock() { + // TODO 自动生成的构造函数存根 + } + @Override + public String toString() { + if(number>0) { + return "DATA"+number+":"+size; + }else { + return "ACK"+(-number); + } + } + @Override + public int hashCode() { + return Objects.hash(number); + } + @Override + public boolean equals(Object obj) { + if (this == obj) + return true; + if (obj == null) + return false; + if (getClass() != obj.getClass()) + return false; + KLALBBlock other = (KLALBBlock) obj; + return number == other.number; + } + @Override + public int compareTo(KLALBBlock o) { + if(time>o.time) { + return 1; + }else if(time tls=new ArrayList<>(); + private TCPListener tcpl; + public KLALBClient(int port) throws IOException { + tcpl=new TCPListener(port); + tcpl.setCon((s)->{ + KLALBClientProtocol kcp=new KLALBClientProtocol(tls); + try { + kcp.setIn(new BufferedInputStream(s.getInputStream(),8192)); + kcp.setOut(new BufferedOutputStream(s.getOutputStream(),8192)); + kcp.startLocal(); + kcp.runProtocol(); + } catch (IOException e) { + e.printStackTrace(); + } + + }); + } + public void open() throws IOException { + tcpl.open(); + } + public List getTls() { + return tls; + } + public static void main(String[] args) throws IOException { + KLALBClient kc=new KLALBClient(4568); + //Tunnel t= new Tunnel("Test1", "127.0.0.1", 4569); + /*Tunnel t= new Tunnel("Test1", "cn-hz-bgp-1.openfrp.top", 65529); + tls.add(t); + Tunnel t2=new Tunnel("Test2","cn-bj-bgp-3.openfrp.top",65529); + tls.add(t2); + Tunnel t3=new Tunnel("Test3","180.76.147.250",65529); + tls.add(t3); + Tunnel t4=new Tunnel("Test4","cn-sx-xa-bgp-1.openfrp.top",65529); + tls.add(t4); + */ + List tls=kc.getTls(); + Tunnel t5=new Tunnel("Test5","la.afrps.cn",49966); + tls.add(t5); + Tunnel t6=new Tunnel("Test6","sg.afrps.cn",49966); + tls.add(t6); + Tunnel t7=new Tunnel("Test7","ch.afrps.cn",49966); + tls.add(t7); + Tunnel t8=new Tunnel("Test8","sj.afrps.cn",49966); + tls.add(t8); + Tunnel t9=new Tunnel("Test9","frp.104300.xyz",49965); + tls.add(t9); + + Tunnel t11=new Tunnel("Test11","153.36.240.12",65529); + tls.add(t11); + Tunnel t12=new Tunnel("Test12","frp.freefrps.com",49965); + tls.add(t12); + kc.open(); + } + + +} diff --git a/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java b/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java new file mode 100644 index 0000000..372e278 --- /dev/null +++ b/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java @@ -0,0 +1,174 @@ +package org.kne.cloud.network.klalb; + +import java.io.*; +import java.util.Iterator; +import java.util.List; +import java.util.UUID; +import java.util.Vector; + +public class KLALBClientProtocol { + + private KLALBCore klc=new KLALBCore(100); + + + private List tls; + private volatile boolean flag=true; + private InputStream in; + private OutputStream out; + + private Listtcps=new Vector<>(); + private UUID uid=UUID.randomUUID(); + + public KLALBClientProtocol(List tls) { + this.tls=tls; + } + + public void startLocal() { + Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{ + try{ + + while(true) { + byte[]b=new byte[Consts.BLOCKSIZE]; + int size=in.read(b); + if(size==-1) + break; + klc.packDataBlock(b,size); + } + }catch(InterruptedException s) { + + }catch(Exception e) { + e.printStackTrace(); + }finally { + closeALL(); + } + }); + Thread downo=ThreadTool.makeVThreadIfSupport("本地发送线程",()->{ + try{ + while(true) { + out.write(klc.unpackDataBlock()); + out.flush(); + } + }catch(InterruptedException s) { + + }catch(Exception e) { + e.printStackTrace(); + }finally { + closeALL(); + } + }); + upo.start(); + downo.start(); + } + public void handleSocket(TCPConnection tc) throws IOException { + tcps.add(tc); + Thread up=ThreadTool.makeVThreadIfSupport("远程发送线程",()->{ + try{ + while(true) { + klc.sendDataBlock(tc); + } + }catch(InterruptedException s) { + + }catch(Exception e) { + tc.close(); + + } + }); + Thread down=ThreadTool.makeVThreadIfSupport("远程接收线程",()->{ + try{ + while(true) { + klc.receiveDataBlock(tc); + } + }catch(InterruptedException s) { + + }catch(Exception e) { + tc.close(); + up.interrupt(); + } + }); + up.start(); + down.start(); + try { + up.join(); + down.join(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + tcps.remove(tc); + } + public void runProtocol() { + for (int i = 0; i < tls.size(); i++) { + Tunnel tll=tls.get(i); + ThreadTool.makeVThreadIfSupport("隧道监视线程",()->{ + while (flag) { + try { + TCPConnection tc=new TCPConnection(tll); + tc.getDout().writeShort(59649); + tc.getDout().writeLong(uid.getMostSignificantBits()); + tc.getDout().writeLong(uid.getLeastSignificantBits()); + tc.getDout().flush(); + + System.out.println("隧道"+tll+"已连接，可用线路数量:"+ tcps.size()+1); + handleSocket(tc); + + int n=tcps.size(); + System.out.println("隧道"+tll+"已断开，可用线路数量:"+ n); + if(n<=0) { + closeALL(); + System.out.println("连接已断开"); + return; + } + } catch (IOException e) { + } + + try { + Thread.sleep(5000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + }).start(); + } + + + + + } + + public InputStream getIn() { + return in; + } + + public void setIn(InputStream in) { + this.in = in; + } + + public OutputStream getOut() { + return out; + } + + public void setOut(OutputStream out) { + this.out = out; + } + public void closeALL() { + flag=false; + try { + in.close(); + + } catch (IOException e) { + e.printStackTrace(); + } + try { + out.close(); + } catch (IOException e) { + e.printStackTrace(); + } + synchronized (tcps) { + + for (int i = 0; i < tcps.size(); i++) { + TCPConnection tll=tcps.get(i); + tll.close(); + } + } + klc.close(); + } +} diff --git a/src/org/kne/cloud/network/klalb/KLALBCore.java b/src/org/kne/cloud/network/klalb/KLALBCore.java new file mode 100644 index 0000000..3c9f9d7 --- /dev/null +++ b/src/org/kne/cloud/network/klalb/KLALBCore.java @@ -0,0 +1,214 @@ +package org.kne.cloud.network.klalb; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.TreeMap; +import java.util.Vector; +import java.util.WeakHashMap; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; + +public class KLALBCore { + private volatile long inputcount = 1; + private volatile long outputcount = 1; + private Set inputcache = Collections.synchronizedSet(new HashSet<>()); + private List outputcache = Collections.synchronizedList(new ArrayList<>()); + + private BlockingQueue kack = new LinkedBlockingQueue<>(); + + + private volatile boolean inlocal=true; + + private volatile boolean close = false; + + + private int cacheblocks; + public KLALBCore(int cachesize) { + cacheblocks=cachesize; + } + public void packDataBlock(byte[] b, int size) throws InterruptedException { + + if (close) + throw new InterruptedException(); + while (outputcache.size() > cacheblocks) { + if (close) + Thread.currentThread().interrupt(); + Thread.sleep(1); + } + outputcache.add(new KLALBBlock(b, size, outputcount++)); + + } + + public byte[] unpackDataBlock() throws InterruptedException { + inlocal=false; + try { + if (close) + throw new InterruptedException(); + byte[] b = null; + while (true) { + synchronized (inputcache) { + Iterator klb = inputcache.iterator(); + while (klb.hasNext()) { + KLALBBlock klalbBlock = (KLALBBlock) klb.next(); + if (klalbBlock.number < inputcount) { + klb.remove(); + } else if (klalbBlock.number == inputcount) { + b = klalbBlock.data; + klb.remove(); + } + } + } + if (b != null) { + System.out.println("\tPROCESS:" + inputcache.size()); + inputcount++; + return b; + } + if (close) + Thread.currentThread().interrupt(); + Thread.sleep(1); + + } + }finally { + inlocal=true; + } + } + + + public void sendDataBlock(TCPConnection out) throws IOException, InterruptedException { + if (close) + throw new InterruptedException(); + while (kack.isEmpty() && outputcache.isEmpty()) { + if (close) + Thread.currentThread().interrupt(); + Thread.sleep(1); + } + KLALBBlock kx = kack.poll(); + if (kx != null) { + kx.time+=System.nanoTime(); + send0(out, kx); + } else { + KLALBBlock ks = null; + synchronized (outputcache) { + for (int i = 0; i < outputcache.size(); i++) { + + KLALBBlock kd = outputcache.get(i); + if (kd.time == -1) { + kd.time = System.nanoTime(); + kd.thread = Thread.currentThread(); + ks = kd; + } else { + if (kd.thread.isAlive()) { + long timex = (System.nanoTime() - kd.time) / 1000000; + if (timex > 10000) { + System.out.println("超时重传："+kd); + kd.time = System.nanoTime(); + kd.thread = Thread.currentThread(); + ks = kd; + } + } else { + System.out.println("掉线重传："+kd); + kd.time = System.nanoTime(); + kd.thread = Thread.currentThread(); + ks = kd; + } + } + if (ks != null) { + break; + } + } + } + if (ks != null) { + send0(out, ks); + // System.out.println(outputcache.size()); + } + } + + } + + public void receiveDataBlock(TCPConnection in) throws IOException, InterruptedException { + if (close) + throw new InterruptedException(); + KLALBBlock x = receive0(in); + if (x.number > 0) { + try { + KLALBBlock klb=new KLALBBlock(null, 0, -x.number); + klb.time=x.time-System.nanoTime(); + kack.put(klb); + } catch (InterruptedException e) { + e.printStackTrace(); + } + if (x.number >= inputcount) { + + while(inputcache.size()>(2*cacheblocks)&&inlocal) { + if (close) + Thread.currentThread().interrupt(); + Thread.sleep(1); + } + + inputcache.add(x); + } + + } else { + long v = -x.number; + outputcache.removeIf((b) -> { + return b.number == v; + }); + } + } + + private void send0(TCPConnection tcp, KLALBBlock kd) throws IOException { + DataOutputStream out=tcp.getDout(); + synchronized (out) { + out.writeLong(kd.number); + if (kd.number > 0) { + out.writeLong(System.nanoTime()); + out.writeInt(kd.size); + out.write(kd.data, 0, kd.size); + }else { + out.writeLong(kd.time); + } + out.flush(); + } + System.out.println("SEND:" + kd); + } + + private KLALBBlock receive0(TCPConnection tcp) throws IOException { + KLALBBlock kb = new KLALBBlock(); + DataInputStream in=tcp.getDin(); + synchronized (in) { + kb.number = in.readLong(); + kb.time = in.readLong(); + + + + if (kb.number > 0) { + kb.size = in.readInt(); + kb.data = new byte[kb.size]; + in.readFully(kb.data); + }else{ + long dela=(System.nanoTime()-kb.time)/2000000; + //System.out.println(tcp.getTunnel()+" "+dela); + } + } + System.out.println("RECEIVE:" + kb); + return kb; + } + + public void close() { + close = true; + } +} diff --git a/src/org/kne/cloud/network/klalb/KLALBServer.java b/src/org/kne/cloud/network/klalb/KLALBServer.java new file mode 100644 index 0000000..5319d0e --- /dev/null +++ b/src/org/kne/cloud/network/klalb/KLALBServer.java @@ -0,0 +1,57 @@ +package org.kne.cloud.network.klalb; + +import java.io.DataInputStream; +import java.io.IOException; +import java.net.ConnectException; +import java.net.Socket; +import java.util.HashMap; +import java.util.UUID; +import java.util.WeakHashMap; + +public class KLALBServer { + WeakHashMap whm=new WeakHashMap<>(); + public KLALBServer(int port) throws IOException { + TCPListener tcpl=new TCPListener(port); + tcpl.setCon((s)->{ + try { + //s.setSoTimeout(10000); + DataInputStream din=new DataInputStream(s.getInputStream()); + int val=din.readShort()&0xffff; + if(val!=59649) { + return; + } + UUID uid=new UUID(din.readLong(),din.readLong()); + System.out.println(uid); + KLALBServerProtocol nx = null; + if(whm.containsKey(uid)) { + nx=whm.get(uid); + }else { + nx=new KLALBServerProtocol(); + Socket soc=new Socket("192.168.1.233",8444); + nx.setOut(soc.getOutputStream()); + nx.setIn(soc.getInputStream()); + nx.startLocal(); + whm.put(uid, nx); + + } + nx.handleSocket(new TCPConnection(null,s)); + }catch(ConnectException e) { + System.out.println("连接本地服务失败，请检查你的服务程序"); + }catch(IOException e) { + e.printStackTrace(); + }finally { + try { + s.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + }); + tcpl.open(); + + } + + public static void main(String[] args) throws IOException { + KLALBServer kc=new KLALBServer(4569); + } +} diff --git a/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java b/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java new file mode 100644 index 0000000..282366b --- /dev/null +++ b/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java @@ -0,0 +1,129 @@ +package org.kne.cloud.network.klalb; + +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; +import java.util.*; +public class KLALBServerProtocol { + private Listtls=new Vector<>(); + + private KLALBCore klc=new KLALBCore(100); + + private InputStream in; + private OutputStream out; + public void startLocal() { + + Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{ + try{ + + while(true) { + byte[]b=new byte[Consts.BLOCKSIZE]; + int size=in.read(b); + if(size==-1) + break; + klc.packDataBlock(b,size); + } + }catch(InterruptedException s) { + + }catch(Exception e) { + e.printStackTrace(); + }finally { + closeALL(); + } + }); + Thread downo=ThreadTool.makeVThreadIfSupport("本地发送线程",()->{ + try{ + while(true) { + + out.write(klc.unpackDataBlock()); + out.flush(); + } + }catch(InterruptedException s) { + + }catch(Exception e) { + e.printStackTrace(); + }finally { + closeALL(); + } + }); + upo.start(); + downo.start(); + } + public void handleSocket(TCPConnection s) throws IOException { + tls.add(s); + Thread up=ThreadTool.makeVThreadIfSupport("远程发送线程",()->{ + try{ + while(true) { + klc.sendDataBlock(s); + } + }catch(InterruptedException s1) { + + }catch(Exception e) { + } + }); + Thread down=ThreadTool.makeVThreadIfSupport("远程接收线程",()->{ + try{ + while(true) { + klc.receiveDataBlock(s); + } + }catch(InterruptedException s1) { + + }catch(Exception e) { + up.interrupt(); + } + }); + + up.start(); + down.start(); + try { + up.join(); + down.join(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + tls.remove(s); + int n=tls.size(); + if(n<=0) { + closeALL(); + System.out.println("连接已关闭"); + } + } + public InputStream getIn() { + return in; + } + + public void setIn(InputStream in) { + this.in = in; + } + + public OutputStream getOut() { + return out; + } + + public void setOut(OutputStream out) { + this.out = out; + } + public void closeALL() { + try { + in.close(); + + } catch (IOException e) { + e.printStackTrace(); + } + try { + out.close(); + } catch (IOException e) { + e.printStackTrace(); + } + for (int i = 0; i < tls.size(); i++) { + TCPConnection tll=tls.get(i); + tll.close(); + } + klc.close(); + } +} diff --git a/src/org/kne/cloud/network/klalb/TCPConnection.java b/src/org/kne/cloud/network/klalb/TCPConnection.java new file mode 100644 index 0000000..986d757 --- /dev/null +++ b/src/org/kne/cloud/network/klalb/TCPConnection.java @@ -0,0 +1,72 @@ +package org.kne.cloud.network.klalb; + +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.net.Socket; +import java.net.UnknownHostException; + +public class TCPConnection { + + private Tunnel tunnel; + private Socket connect; +private DataInputStream din; +public Tunnel getTunnel() { + return tunnel; +} +private DataOutputStream dout; +private long delay; +public Socket getConnect() { + return connect; +} +public DataInputStream getDin() { + return din; +} +public DataOutputStream getDout() { + return dout; +} +public void close() { + // TODO 自动生成的方法存根 + try { + if(din!=null) + din.close(); + } catch (IOException e1) { + // TODO 自动生成的 catch 块 + e1.printStackTrace(); + } + try { + if(dout!=null) + dout.close(); + } catch (IOException e1) { + // TODO 自动生成的 catch 块 + e1.printStackTrace(); + } + try { + if(connect!=null) + connect.close(); + } catch (IOException e) { + // TODO 自动生成的 catch 块 + e.printStackTrace(); + } + connect=null; +} +public long getDelay() { + return delay; +} +public void setDelay(long delay) { + this.delay = delay; +} +public TCPConnection(Tunnel t) throws UnknownHostException, IOException { + this(t,t.connectClientSocket()); +} +public TCPConnection(Tunnel t,Socket soc) throws IOException { + connect=soc; + tunnel=t; + //connect.setSoTimeout(10000); + din=new DataInputStream(new BufferedInputStream( connect.getInputStream(),65536)); + dout=new DataOutputStream(new BufferedOutputStream (connect.getOutputStream(),65536)); + +} +} diff --git a/src/org/kne/cloud/network/klalb/TCPListener.java b/src/org/kne/cloud/network/klalb/TCPListener.java new file mode 100644 index 0000000..48121b1 --- /dev/null +++ b/src/org/kne/cloud/network/klalb/TCPListener.java @@ -0,0 +1,65 @@ +package org.kne.cloud.network.klalb; + +import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.function.Consumer; + +public class TCPListener { + private int port; + private ServerSocket servers; + private volatile boolean flag=false; + + private Consumercon; + private Runnable r=new Runnable() { + @Override + public void run() { + while(flag){ + try { + Socket soce=servers.accept(); + ThreadTool.makeVThreadIfSupport("端口监听线程",()->{ + con.accept(soce); + }).start(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + }; + public TCPListener(int port) throws IOException { + this.port=port; + } + + public void open() throws IOException { + flag=true; + servers=new ServerSocket(port); + new Thread(r).start(); + } + public void close() { + flag=false; + if(servers!=null) { + try { + servers.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + + + public int getPort() { + return port; + } + public void setPort(int port) { + this.port = port; + } + + public Consumer getCon() { + return con; + } + + public void setCon(Consumer con) { + this.con = con; + } + +} diff --git a/src/org/kne/cloud/network/klalb/ThreadTool.java b/src/org/kne/cloud/network/klalb/ThreadTool.java new file mode 100644 index 0000000..913c2b6 --- /dev/null +++ b/src/org/kne/cloud/network/klalb/ThreadTool.java @@ -0,0 +1,20 @@ +package org.kne.cloud.network.klalb; + +public class ThreadTool { + public static boolean first=true; + public static boolean forceD=false; + public static Thread makeVThreadIfSupport(String name,Runnable r) { + if(forceD) + return new Thread(r, name); + try { + return Thread.ofVirtual().name(name).unstarted(r); + }catch(Throwable e) { + if(first) { + System.out.println("请使用java19以上版本以提高性能！"); + first=false; + } + return new Thread(r, name); + } + } + +} diff --git a/src/org/kne/cloud/network/klalb/Tunnel.java b/src/org/kne/cloud/network/klalb/Tunnel.java new file mode 100644 index 0000000..a51cbe8 --- /dev/null +++ b/src/org/kne/cloud/network/klalb/Tunnel.java @@ -0,0 +1,52 @@ +package org.kne.cloud.network.klalb; + +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.net.Socket; +import java.net.UnknownHostException; + +public class Tunnel { + private String name; + private String ip; + private int port; + + public Tunnel(String name, String ip, int port) { + super(); + this.name = name; + this.ip = ip; + this.port = port; + } + public String getName() { + return name; + } + public void setName(String name) { + this.name = name; + } + public String getIp() { + return ip; + } + public void setIp(String ip) { + this.ip = ip; + } + public int getPort() { + return port; + } + public void setPort(int port) { + this.port = port; + } + + @Override + public String toString() { + return name+"$"+ip+"$"+port; + } + public Socket connectClientSocket() throws UnknownHostException, IOException { + return new Socket(ip, port); + } + + + + +}