From 642c1971430a5fa785d50d1a58652e77cb0e68d7 Mon Sep 17 00:00:00 2001 From: Administrator Date: Sat, 26 Nov 2022 16:38:09 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=88=E5=B9=B6IO=E7=BA=BF=E7=A8=8B=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cloud/network/klalb/IOThreadManager.java | 132 ++++++++++++++++++ .../kne/cloud/network/klalb/KLALBClient.java | 43 +++++- .../network/klalb/KLALBClientProtocol.java | 63 ++------- .../kne/cloud/network/klalb/KLALBCore.java | 24 +++- .../kne/cloud/network/klalb/KLALBServer.java | 15 +- .../network/klalb/KLALBServerProtocol.java | 25 +++- .../cloud/network/klalb/TCPConnection.java | 2 +- 7 files changed, 235 insertions(+), 69 deletions(-) create mode 100644 src/org/kne/cloud/network/klalb/IOThreadManager.java diff --git a/src/org/kne/cloud/network/klalb/IOThreadManager.java b/src/org/kne/cloud/network/klalb/IOThreadManager.java new file mode 100644 index 0000000..d99881c --- /dev/null +++ b/src/org/kne/cloud/network/klalb/IOThreadManager.java @@ -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 Listtcps=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 getTcps() { + return tcps; + } + public boolean isClosed() { + return closed; + } +} diff --git a/src/org/kne/cloud/network/klalb/KLALBClient.java b/src/org/kne/cloud/network/klalb/KLALBClient.java index 6a4bc93..e9a505c 100644 --- a/src/org/kne/cloud/network/klalb/KLALBClient.java +++ b/src/org/kne/cloud/network/klalb/KLALBClient.java @@ -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(); diff --git a/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java b/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java index 372e278..af97a52 100644 --- a/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java +++ b/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java @@ -7,21 +7,13 @@ 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 Listtcps=new Vector<>(); private InputStream in; private OutputStream out; - - private Listtcps=new Vector<>(); - private UUID uid=UUID.randomUUID(); - - public KLALBClientProtocol(List 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 getTcps() { + return tcps; + } + + public boolean isClosed() { + return closed; + } + } diff --git a/src/org/kne/cloud/network/klalb/KLALBCore.java b/src/org/kne/cloud/network/klalb/KLALBCore.java index 529690c..530c74e 100644 --- a/src/org/kne/cloud/network/klalb/KLALBCore.java +++ b/src/org/kne/cloud/network/klalb/KLALBCore.java @@ -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 inputcache = Collections.synchronizedSet(new HashSet<>()); private List 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(); + } + } } } diff --git a/src/org/kne/cloud/network/klalb/KLALBServer.java b/src/org/kne/cloud/network/klalb/KLALBServer.java index 5b58ae1..f2da435 100644 --- a/src/org/kne/cloud/network/klalb/KLALBServer.java +++ b/src/org/kne/cloud/network/klalb/KLALBServer.java @@ -9,11 +9,11 @@ import java.util.UUID; import java.util.WeakHashMap; public class KLALBServer { - WeakHashMap whm=new WeakHashMap<>(); + WeakHashMap 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) { diff --git a/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java b/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java index 282366b..3387bbd 100644 --- a/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java +++ b/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java @@ -10,12 +10,15 @@ 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 Listtcps=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 getTcps() { + return tcps; + } + public boolean isClosed() { + return closed; + } + } diff --git a/src/org/kne/cloud/network/klalb/TCPConnection.java b/src/org/kne/cloud/network/klalb/TCPConnection.java index c1f18b7..50a62c2 100644 --- a/src/org/kne/cloud/network/klalb/TCPConnection.java +++ b/src/org/kne/cloud/network/klalb/TCPConnection.java @@ -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();