From 4f2b416ac3aa6843c24c71f59ee2103e12632cc5 Mon Sep 17 00:00:00 2001 From: Administrator Date: Sun, 27 Nov 2022 06:36:27 +0800 Subject: [PATCH] =?UTF-8?q?=E6=80=A7=E8=83=BD=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .classpath | 6 +- .settings/org.eclipse.jdt.core.prefs | 8 +- .../cloud/network/klalb/IOThreadManager.java | 69 +++++---- .../kne/cloud/network/klalb/KLALBClient.java | 23 ++- .../network/klalb/KLALBClientProtocol.java | 139 ----------------- .../kne/cloud/network/klalb/KLALBCore.java | 77 +++++++--- .../kne/cloud/network/klalb/KLALBServer.java | 5 +- .../network/klalb/KLALBServerProtocol.java | 140 ------------------ .../cloud/network/klalb/TCPConnection.java | 30 +--- .../kne/cloud/network/klalb/ThreadTool.java | 4 +- 10 files changed, 129 insertions(+), 372 deletions(-) delete mode 100644 src/org/kne/cloud/network/klalb/KLALBClientProtocol.java delete mode 100644 src/org/kne/cloud/network/klalb/KLALBServerProtocol.java diff --git a/.classpath b/.classpath index ae62d39..af6f9dc 100644 --- a/.classpath +++ b/.classpath @@ -1,10 +1,6 @@ - - - - - + diff --git a/.settings/org.eclipse.jdt.core.prefs b/.settings/org.eclipse.jdt.core.prefs index 17d114f..c59d0c6 100644 --- a/.settings/org.eclipse.jdt.core.prefs +++ b/.settings/org.eclipse.jdt.core.prefs @@ -1,9 +1,9 @@ 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.targetPlatform=1.8 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve -org.eclipse.jdt.core.compiler.compliance=18 +org.eclipse.jdt.core.compiler.compliance=1.8 org.eclipse.jdt.core.compiler.debug.lineNumber=generate org.eclipse.jdt.core.compiler.debug.localVariable=generate org.eclipse.jdt.core.compiler.debug.sourceFile=generate @@ -11,5 +11,5 @@ 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 +org.eclipse.jdt.core.compiler.release=disabled +org.eclipse.jdt.core.compiler.source=1.8 diff --git a/src/org/kne/cloud/network/klalb/IOThreadManager.java b/src/org/kne/cloud/network/klalb/IOThreadManager.java index d99881c..5f6c945 100644 --- a/src/org/kne/cloud/network/klalb/IOThreadManager.java +++ b/src/org/kne/cloud/network/klalb/IOThreadManager.java @@ -3,6 +3,7 @@ package org.kne.cloud.network.klalb; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.net.SocketException; import java.util.List; import java.util.Vector; @@ -11,11 +12,10 @@ public class IOThreadManager { private Listtcps=new Vector<>(); - private InputStream in; - private OutputStream out; + private TCPConnection local; - private volatile boolean closed=true; + private volatile boolean open=true; public void startLocal() { Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{ @@ -23,30 +23,32 @@ public class IOThreadManager { while(true) { byte[]b=new byte[Consts.BLOCKSIZE]; - int size=in.read(b); + int size=local.getDin().read(b); if(size==-1) break; klc.packDataBlock(b,size); } + klc.waitForRemote(); }catch(InterruptedException s) { + }catch(SocketException se) { }catch(Exception e) { e.printStackTrace(); }finally { - closeALL(); + closeRemote(); } }); Thread downo=ThreadTool.makeVThreadIfSupport("本地发送线程",()->{ try{ while(true) { - out.write(klc.unpackDataBlock()); - out.flush(); + local.getDout().write(klc.unpackDataBlock()); + local.getDout().flush(); } }catch(InterruptedException s) { }catch(Exception e) { e.printStackTrace(); }finally { - closeALL(); + closeRemote(); } }); upo.start(); @@ -80,7 +82,6 @@ public class IOThreadManager { up.start(); down.start(); try { - up.join(); down.join(); } catch (InterruptedException e) { e.printStackTrace(); @@ -89,23 +90,39 @@ public class IOThreadManager { tcps.remove(s); } } - public InputStream getIn() { - return in; + + public void closeRemote() { + open=false; + + for (int i = 0; i < tcps.size(); i++) { + TCPConnection tll=tcps.get(i); + tll.close(); + } + klc.closeRemote(); } - - public void setIn(InputStream in) { - this.in = in; + + public void closeLocal() { + open=false; + local.close(); + klc.closeLocal(); } - - public OutputStream getOut() { - return out; + + public TCPConnection getLocal() { + return local; } - - public void setOut(OutputStream out) { - this.out = out; + public void setLocal(TCPConnection local) { + this.local = local; } + /* public void closeALL() { - closed=false; + open=false; + + for (int i = 0; i < tcps.size(); i++) { + TCPConnection tll=tcps.get(i); + tll.close(); + } + klc.closeRemote(); + klc.closeLocal(); try { in.close(); @@ -117,16 +134,12 @@ public class IOThreadManager { } 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; + public boolean isOpen() { + return open; } } diff --git a/src/org/kne/cloud/network/klalb/KLALBClient.java b/src/org/kne/cloud/network/klalb/KLALBClient.java index e9a505c..1568991 100644 --- a/src/org/kne/cloud/network/klalb/KLALBClient.java +++ b/src/org/kne/cloud/network/klalb/KLALBClient.java @@ -6,6 +6,8 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; public class KLALBClient { private List tls=new ArrayList<>(); @@ -16,8 +18,7 @@ public class KLALBClient { IOThreadManager kcp=new IOThreadManager(); UUID uid=UUID.randomUUID(); try { - kcp.setIn(new BufferedInputStream(s.getInputStream(),8192)); - kcp.setOut(new BufferedOutputStream(s.getOutputStream(),8192)); + kcp.setLocal(new TCPConnection(null, s)); kcp.startLocal(); runProtocol(kcp,uid); } catch (IOException e) { @@ -27,28 +28,40 @@ public class KLALBClient { }); } public void runProtocol(IOThreadManager kcp, UUID uid) { + AtomicBoolean b=new AtomicBoolean(true); + AtomicInteger aig=new AtomicInteger(0); for (int i = 0; i < tls.size(); i++) { Tunnel tll=tls.get(i); ThreadTool.makeVThreadIfSupport("隧道监视线程",()->{ - while (kcp.isClosed()) { + while (kcp.isOpen()) { try { TCPConnection tc=new TCPConnection(tll); tc.getDout().writeShort(59649); tc.getDout().writeLong(uid.getMostSignificantBits()); tc.getDout().writeLong(uid.getLeastSignificantBits()); tc.getDout().flush(); - + aig.incrementAndGet(); 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(); + kcp.closeRemote(); + kcp.closeLocal(); System.out.println("连接已断开"); return; } } catch (IOException e) { + if(aig.get()<=0) { + kcp.closeRemote(); + kcp.closeLocal(); + if(b.get()) { + System.out.println("连接失败"); + b.set(false); + } + return; + } } try { diff --git a/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java b/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java deleted file mode 100644 index af97a52..0000000 --- a/src/org/kne/cloud/network/klalb/KLALBClientProtocol.java +++ /dev/null @@ -1,139 +0,0 @@ -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 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 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 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(); - } - synchronized (tcps) { - - 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/KLALBCore.java b/src/org/kne/cloud/network/klalb/KLALBCore.java index 530c74e..175f850 100644 --- a/src/org/kne/cloud/network/klalb/KLALBCore.java +++ b/src/org/kne/cloud/network/klalb/KLALBCore.java @@ -31,23 +31,24 @@ public class KLALBCore { private Set inputcache = Collections.synchronizedSet(new HashSet<>()); private List outputcache = Collections.synchronizedList(new ArrayList<>()); + private Map>acks=Collections.synchronizedMap(new WeakHashMap<>()); + private volatile boolean inlocal=true; - private volatile boolean close = false; - + private volatile boolean closeremote = false; + private volatile boolean closelocal = false; 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) + if (closelocal) throw new InterruptedException(); while (outputcache.size() > cacheblocks) { - if (close) + if (closelocal) Thread.currentThread().interrupt(); Thread.sleep(1); } @@ -58,7 +59,7 @@ public class KLALBCore { public byte[] unpackDataBlock() throws InterruptedException { inlocal=false; try { - if (close) + if (closelocal) throw new InterruptedException(); byte[] b = null; while (true) { @@ -79,7 +80,7 @@ public class KLALBCore { inputcount++; return b; } - if (close) + if (closelocal) Thread.currentThread().interrupt(); Thread.sleep(1); @@ -91,13 +92,20 @@ public class KLALBCore { public void sendDataBlock(TCPConnection out) throws IOException, InterruptedException { - if (close) + if (closeremote) throw new InterruptedException(); - while ( outputcache.isEmpty()) { - if (close) + BlockingQueue bqk=acks.get(out); + while ((bqk==null||bqk.isEmpty())&& outputcache.isEmpty()) { + if (closeremote) Thread.currentThread().interrupt(); Thread.sleep(1); } + + if(bqk!=null&&bqk.size()>0) { + KLALBBlock klb=bqk.poll(); + if(klb!=null) + send0(out, klb); + }else { KLALBBlock ks = null; synchronized (outputcache) { for (int i = 0; i < outputcache.size(); i++) { @@ -131,30 +139,41 @@ public class KLALBCore { if (ks != null) { send0(out, ks); // System.out.println(outputcache.size()); + }else { + Thread.sleep(1); + } } - } public void receiveDataBlock(TCPConnection in) throws IOException, InterruptedException { - if (close) + if (closeremote) throw new InterruptedException(); KLALBBlock x = receive0(in); if (x.number > 0) { KLALBBlock klb=new KLALBBlock(null, 0, -x.number); - exec.execute( ()->{ + BlockingQueue bq=acks.get(in); + if(bq==null) { + BlockingQueue bqt=new LinkedBlockingQueue<>(); + bqt.add(klb); + acks.put(in,bqt); + }else { + acks.get(in).add(klb); + } + /*ThreadTool.makeVThreadIfSupport("TACK", ()->{ try { send0(in, klb); } catch (IOException e) { e.printStackTrace(); }finally { } - }); + }).start();*/ + if (x.number >= inputcount) { while(inputcache.size()>(2*cacheblocks)&&inlocal) { - if (close) + if (closeremote) Thread.currentThread().interrupt(); Thread.sleep(1); } @@ -178,9 +197,8 @@ public class KLALBCore { out.writeInt(kd.size); out.write(kd.data, 0, kd.size); } + out.flush(); - - tcp.flush(); } System.out.println("SEND:" + kd); } @@ -200,12 +218,27 @@ public class KLALBCore { return kb; } - public void close() { - close = true; - exec.shutdown(); + public void closeRemote() { + closeremote = true; + outputcache.clear(); } - public void waitForEnding() { - while(outputcache.size()>0||inputcache.size()>0) { + public void closeLocal() { + closelocal = true; + inputcache.clear(); + } + + + public void waitForRemote() { + while(outputcache.size()>0) { + try { + Thread.sleep(1); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + } + public void waitForLocal() { + while(inputcache.size()>0) { try { Thread.sleep(1); } catch (InterruptedException e) { diff --git a/src/org/kne/cloud/network/klalb/KLALBServer.java b/src/org/kne/cloud/network/klalb/KLALBServer.java index f2da435..964673b 100644 --- a/src/org/kne/cloud/network/klalb/KLALBServer.java +++ b/src/org/kne/cloud/network/klalb/KLALBServer.java @@ -28,8 +28,7 @@ public class KLALBServer { }else { nx=new IOThreadManager(); Socket soc=new Socket("192.168.1.233",8444); - nx.setOut(soc.getOutputStream()); - nx.setIn(soc.getInputStream()); + nx.setLocal(new TCPConnection(null, soc)); nx.startLocal(); whm.put(uid, nx); @@ -37,7 +36,7 @@ public class KLALBServer { nx.handleSocket(new TCPConnection(null,s)); int n=nx.getTcps().size(); if(n<=0) { - nx.closeALL(); + nx.closeLocal(); System.out.println("连接已关闭"); } }catch(ConnectException e) { diff --git a/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java b/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java deleted file mode 100644 index 3387bbd..0000000 --- a/src/org/kne/cloud/network/klalb/KLALBServerProtocol.java +++ /dev/null @@ -1,140 +0,0 @@ -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 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); - 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(); - } - tcps.remove(s); - int n=tcps.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() { - 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/TCPConnection.java b/src/org/kne/cloud/network/klalb/TCPConnection.java index 50a62c2..bd65b7e 100644 --- a/src/org/kne/cloud/network/klalb/TCPConnection.java +++ b/src/org/kne/cloud/network/klalb/TCPConnection.java @@ -34,7 +34,6 @@ public class TCPConnection { } public void close() { - // TODO 自动生成的方法存根 try { if (din != null) din.close(); @@ -70,36 +69,19 @@ public class TCPConnection { 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)); - ThreadTool.makeVThreadIfSupport("FLUSH", () -> { - try { - while (true) { - Thread.sleep(10); - if(flush) { - synchronized (dout) { - dout.flush(); - - } - flush=false; - } - } - } catch (IOException e) { - e.printStackTrace(); - } catch (InterruptedException e) { - e.printStackTrace(); - } - }).start(); } - private volatile boolean flush = false; - - public void flush() throws IOException { - flush = true; + public boolean isOpen() { + if(connect==null) { + return false; + } + return !connect.isClosed(); } + } diff --git a/src/org/kne/cloud/network/klalb/ThreadTool.java b/src/org/kne/cloud/network/klalb/ThreadTool.java index 913c2b6..d28cf69 100644 --- a/src/org/kne/cloud/network/klalb/ThreadTool.java +++ b/src/org/kne/cloud/network/klalb/ThreadTool.java @@ -6,8 +6,8 @@ public class ThreadTool { public static Thread makeVThreadIfSupport(String name,Runnable r) { if(forceD) return new Thread(r, name); - try { - return Thread.ofVirtual().name(name).unstarted(r); + try { return new Thread(r, name); + //return Thread.ofVirtual().name(name).unstarted(r); }catch(Throwable e) { if(first) { System.out.println("请使用java19以上版本以提高性能！");