KNEMC/KLALB
1
2
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

KLALB

Browse Source
This commit is contained in:
Administrator
2022-11-25 21:06:10 +08:00
commit 0b5ff2465b
16 changed files with 956 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/org/kne/cloud/network/klalb/Consts.java
+5
View File
@@ -0,0 +1,5 @@
package org.kne.cloud.network.klalb;
public class Consts {
public static int BLOCKSIZE=65536;
}
src/org/kne/cloud/network/klalb/KLALBBlock.java
+57
View File
@@ -0,0 +1,57 @@
package org.kne.cloud.network.klalb;
import java.util.Arrays;
import java.util.Objects;
public class KLALBBlock implements Comparable<KLALBBlock>{
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<o.time){
return -1;
}else {
return 0;
}
}
}
src/org/kne/cloud/network/klalb/KLALBClient.java
+66
View File
@@ -0,0 +1,66 @@
package org.kne.cloud.network.klalb;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
public class KLALBClient {
private List<Tunnel> 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<Tunnel> 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<Tunnel> 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();
}
}
src/org/kne/cloud/network/klalb/KLALBClientProtocol.java
+174
View File
@@ -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<Tunnel> tls;
private volatile boolean flag=true;
private InputStream in;
private OutputStream out;
private List<TCPConnection>tcps=new Vector<>();
private UUID uid=UUID.randomUUID();
public KLALBClientProtocol(List<Tunnel> 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();
}
}
src/org/kne/cloud/network/klalb/KLALBCore.java
+214
View File
@@ -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<KLALBBlock> inputcache = Collections.synchronizedSet(new HashSet<>());
private List<KLALBBlock> outputcache = Collections.synchronizedList(new ArrayList<>());
private BlockingQueue<KLALBBlock> 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<KLALBBlock> 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;
}
}
src/org/kne/cloud/network/klalb/KLALBServer.java
+57
View File
@@ -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<UUID, KLALBServerProtocol> 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);
}
}
src/org/kne/cloud/network/klalb/KLALBServerProtocol.java
+129
View File
@@ -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 List<TCPConnection>tls=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();
}
}
src/org/kne/cloud/network/klalb/TCPConnection.java
+72
View File
@@ -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));
}
}
src/org/kne/cloud/network/klalb/TCPListener.java
+65
View File
@@ -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 Consumer<Socket>con;
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<Socket> getCon() {
return con;
}
public void setCon(Consumer<Socket> con) {
this.con = con;
}
}
src/org/kne/cloud/network/klalb/ThreadTool.java
+20
View File
@@ -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);
}
}
}
src/org/kne/cloud/network/klalb/Tunnel.java
+52
View File
@@ -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);
}
}