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

合并IO线程管理类

Browse Source
This commit is contained in:
Administrator
2022-11-26 16:38:09 +08:00
parent 479204919b
commit 642c197143
7 changed files with 235 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/org/kne/cloud/network/klalb/IOThreadManager.java
+132
View File
@@ -0,0 +1,132 @@
package org.kne.cloud.network.klalb;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;
import java.util.Vector;
public class IOThreadManager {
private KLALBCore klc=new KLALBCore(100);
private List<TCPConnection>tcps=new Vector<>();
private InputStream in;
private OutputStream out;
private volatile boolean closed=true;
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 {
tcps.add(s);
try {
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();
}
}finally {
tcps.remove(s);
}
}
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() {
closed=false;
try {
in.close();
} catch (IOException e) {
e.printStackTrace();
}
try {
out.close();
} catch (IOException e) {
e.printStackTrace();
}
for (int i = 0; i < tcps.size(); i++) {
TCPConnection tll=tcps.get(i);
tll.close();
}
klc.close();
}
public List<TCPConnection> getTcps() {
return tcps;
}
public boolean isClosed() {
return closed;
}
}
src/org/kne/cloud/network/klalb/KLALBClient.java
+41 -2
View File
@@ -13,17 +13,56 @@ public class KLALBClient {
public KLALBClient(int port) throws IOException {
tcpl=new TCPListener(port);
tcpl.setCon((s)->{
KLALBClientProtocol kcp=new KLALBClientProtocol(tls);
IOThreadManager kcp=new IOThreadManager();
UUID uid=UUID.randomUUID();
try {
kcp.setIn(new BufferedInputStream(s.getInputStream(),8192));
kcp.setOut(new BufferedOutputStream(s.getOutputStream(),8192));
kcp.startLocal();
kcp.runProtocol();
runProtocol(kcp,uid);
} catch (IOException e) {
e.printStackTrace();
}
});
}
public void runProtocol(IOThreadManager kcp, UUID uid) {
for (int i = 0; i < tls.size(); i++) {
Tunnel tll=tls.get(i);
ThreadTool.makeVThreadIfSupport("隧道监视线程",()->{
while (kcp.isClosed()) {
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+"已连接，可用线路数量:"+ (kcp.getTcps().size()+1));
kcp.handleSocket(tc);
int n=kcp.getTcps().size();
System.out.println("隧道"+tll+"已断开，可用线路数量:"+ n);
if(n<=0) {
kcp.closeALL();
System.out.println("连接已断开");
return;
}
} catch (IOException e) {
}
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}
public void open() throws IOException {
tcpl.open();
src/org/kne/cloud/network/klalb/KLALBClientProtocol.java
+14 -49
View File
@@ -7,21 +7,13 @@ 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 List<TCPConnection>tcps=new Vector<>();
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;
}
private volatile boolean closed=true;
public void startLocal() {
Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{
@@ -95,44 +87,9 @@ public class KLALBClientProtocol {
}
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;
@@ -150,7 +107,7 @@ public class KLALBClientProtocol {
this.out = out;
}
public void closeALL() {
flag=false;
closed=false;
try {
in.close();
@@ -171,4 +128,12 @@ public class KLALBClientProtocol {
}
klc.close();
}
public List<TCPConnection> getTcps() {
return tcps;
}
public boolean isClosed() {
return closed;
}
}
src/org/kne/cloud/network/klalb/KLALBCore.java
+19 -5
View File
@@ -20,7 +20,10 @@ import java.util.WeakHashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class KLALBCore {
private volatile long inputcount = 1;
@@ -28,17 +31,17 @@ public class KLALBCore {
private Set<KLALBBlock> inputcache = Collections.synchronizedSet(new HashSet<>());
private List<KLALBBlock> outputcache = Collections.synchronizedList(new ArrayList<>());
private volatile boolean inlocal=true;
private volatile boolean close = false;
private int cacheblocks;
private volatile int cacheblocks;
public KLALBCore(int cachesize) {
cacheblocks=cachesize;
}
private ExecutorService exec=Executors.newCachedThreadPool();
public void packDataBlock(byte[] b, int size) throws InterruptedException {
if (close)
@@ -139,13 +142,14 @@ public class KLALBCore {
if (x.number > 0) {
KLALBBlock klb=new KLALBBlock(null, 0, -x.number);
ThreadTool.makeVThreadIfSupport("ACK", ()->{
exec.execute( ()->{
try {
send0(in, klb);
} catch (IOException e) {
e.printStackTrace();
}finally {
}
}).start();
});
if (x.number >= inputcount) {
@@ -198,5 +202,15 @@ public class KLALBCore {
public void close() {
close = true;
exec.shutdown();
}
public void waitForEnding() {
while(outputcache.size()>0||inputcache.size()>0) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
src/org/kne/cloud/network/klalb/KLALBServer.java
+10 -5
View File
@@ -9,11 +9,11 @@ import java.util.UUID;
import java.util.WeakHashMap;
public class KLALBServer {
WeakHashMap<UUID, KLALBServerProtocol> whm=new WeakHashMap<>();
WeakHashMap<UUID, IOThreadManager> whm=new WeakHashMap<>();
public KLALBServer(int port) throws IOException {
TCPListener tcpl=new TCPListener(port);
tcpl.setCon((s)->{
try {
try {
//s.setSoTimeout(10000);
DataInputStream din=new DataInputStream(s.getInputStream());
int val=din.readShort()&0xffff;
@@ -22,12 +22,12 @@ public class KLALBServer {
}
UUID uid=new UUID(din.readLong(),din.readLong());
System.out.println(uid);
KLALBServerProtocol nx = null;
IOThreadManager nx = null;
if(whm.containsKey(uid)) {
nx=whm.get(uid);
}else {
nx=new KLALBServerProtocol();
Socket soc=new Socket("192.168.1.233",3389);
nx=new IOThreadManager();
Socket soc=new Socket("192.168.1.233",8444);
nx.setOut(soc.getOutputStream());
nx.setIn(soc.getInputStream());
nx.startLocal();
@@ -35,6 +35,11 @@ public class KLALBServer {
}
nx.handleSocket(new TCPConnection(null,s));
int n=nx.getTcps().size();
if(n<=0) {
nx.closeALL();
System.out.println("连接已关闭");
}
}catch(ConnectException e) {
System.out.println("连接本地服务失败，请检查你的服务程序");
}catch(IOException e) {
src/org/kne/cloud/network/klalb/KLALBServerProtocol.java
+18 -7
View File
@@ -10,12 +10,15 @@ 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 List<TCPConnection>tcps=new Vector<>();
private InputStream in;
private OutputStream out;
private volatile boolean closed=true;
public void startLocal() {
Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{
@@ -55,7 +58,7 @@ public class KLALBServerProtocol {
downo.start();
}
public void handleSocket(TCPConnection s) throws IOException {
tls.add(s);
tcps.add(s);
Thread up=ThreadTool.makeVThreadIfSupport("远程发送线程",()->{
try{
while(true) {
@@ -86,8 +89,8 @@ public class KLALBServerProtocol {
} catch (InterruptedException e) {
e.printStackTrace();
}
tls.remove(s);
int n=tls.size();
tcps.remove(s);
int n=tcps.size();
if(n<=0) {
closeALL();
System.out.println("连接已关闭");
@@ -109,6 +112,7 @@ public class KLALBServerProtocol {
this.out = out;
}
public void closeALL() {
closed=false;
try {
in.close();
@@ -120,10 +124,17 @@ public class KLALBServerProtocol {
} catch (IOException e) {
e.printStackTrace();
}
for (int i = 0; i < tls.size(); i++) {
TCPConnection tll=tls.get(i);
for (int i = 0; i < tcps.size(); i++) {
TCPConnection tll=tcps.get(i);
tll.close();
}
klc.close();
}
public List<TCPConnection> getTcps() {
return tcps;
}
public boolean isClosed() {
return closed;
}
}
src/org/kne/cloud/network/klalb/TCPConnection.java
+1 -1
View File
@@ -80,7 +80,7 @@ public class TCPConnection {
ThreadTool.makeVThreadIfSupport("FLUSH", () -> {
try {
while (true) {
Thread.sleep(100);
Thread.sleep(10);
if(flush) {
synchronized (dout) {
dout.flush();