From cc1df750ae044ea13213dd652a959eef7f7ed043 Mon Sep 17 00:00:00 2001 From: Administrator Date: Thu, 16 Mar 2023 16:06:39 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=86=85=E5=AD=98=E6=B3=84?= =?UTF-8?q?=E9=9C=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- KLALB/KNECloud/iplist.json | 20 +++- src/org/kne/cloud/network/klalb/Consts.java | 3 +- .../cloud/network/klalb/IOThreadManager.java | 96 ++++++++++++------- .../kne/cloud/network/klalb/KLALBBlock.java | 2 +- .../kne/cloud/network/klalb/KLALBCore.java | 76 ++++++++++----- .../kne/cloud/network/klalb/KLALBServer.java | 36 +++++-- .../network/klalb/LocalTCPConnection.java | 7 +- .../network/klalb/RemoteTCPConnection.java | 3 + .../cloud/network/klalb/TCPConnection.java | 8 +- src/org/kne/ui/YScrollPane.java | 6 +- 10 files changed, 181 insertions(+), 76 deletions(-) diff --git a/KLALB/KNECloud/iplist.json b/KLALB/KNECloud/iplist.json index 06ec9b6..836089b 100644 --- a/KLALB/KNECloud/iplist.json +++ b/KLALB/KNECloud/iplist.json @@ -1,4 +1,4 @@ -{ +{ "name" : "KNECloud", "services" : [ { "name" : "KNE官网", @@ -12,15 +12,25 @@ "name" : "服务器远程桌面", "protocol" : "RDP", "localaddress" : "127.0.0.1:3389" - } ], - "tunnels" : [ { - "name" : "Openfrp-直连线路", - "remoteaddress" : "127.0.0.1:4566" }, { + "name" : "MC", + "protocol" : "Minecraft", + "localaddress" : "127.0.0.1:25565" + } ], + "tunnels" : [ { "name" : "Openfrp-直连线路", "remoteaddress" : "127.0.0.1:4566", "frpc" : [ "[common]", "server_addr = 180.76.147.250", "server_port = 8120", "tcp_mux = true", "protocol = tcp", "dns_server = 223.5.5.5", "user = bb28264e0abf3bcbec3a2180c5a279b8", "token = I2KMo1HvRxuvurv2", "[ts1xx]", "type = xtcp", "role = visitor", "server_name = zl1", "bind_addr = 127.0.0.1", "bind_port = 4566", "sk = knecloud" ] }, { + "name" : "Openfrp-直连线路", + "remoteaddress" : "127.0.0.1:4566" + }, { + "name" : "Openfrp-直连线路", + "remoteaddress" : "127.0.0.1:4566" + }, { + "name" : "Openfrp-直连线路", + "remoteaddress" : "127.0.0.1:4566" + },{ "name" : "NULL-宿迁联通", "remoteaddress" : "153.36.240.12:65529" }, { diff --git a/src/org/kne/cloud/network/klalb/Consts.java b/src/org/kne/cloud/network/klalb/Consts.java index 942376c..066a209 100644 --- a/src/org/kne/cloud/network/klalb/Consts.java +++ b/src/org/kne/cloud/network/klalb/Consts.java @@ -5,8 +5,9 @@ public class Consts { public static final long PINGTIMENS=1000000000L; public static final int SO_TIMEOUT = 5000; public static final long SN_KEEP = 60000000000L; - public static final int MAX_RESEND = 100; + public static final int MAX_RESEND = 10; public static final long UACK_TIME = 5000000000L; public static final int CONNECT_TIMEOUT=10000; public static final double MAX_QUEUE_TIME=500; + public static final long KEEP_TIME = 60000000000L; } diff --git a/src/org/kne/cloud/network/klalb/IOThreadManager.java b/src/org/kne/cloud/network/klalb/IOThreadManager.java index e6d92c9..7ed8584 100644 --- a/src/org/kne/cloud/network/klalb/IOThreadManager.java +++ b/src/org/kne/cloud/network/klalb/IOThreadManager.java @@ -17,6 +17,7 @@ import java.util.concurrent.BlockingDeque; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import org.kne.cloud.network.mport.ServiceElement; import org.kne.cloud.network.mport.ThreadTool; @@ -35,43 +36,44 @@ public class IOThreadManager { } klc.getLocaltcps().put(tc.getCuid(), tc); - if(klc.getRemotetcps().isEmpty()) { - System.out.println("所有线路已断开，请检查网络"); - Collection c=klc.getLocaltcps().values(); - for (Iterator iterator = c.iterator(); iterator.hasNext();) { - LocalTCPConnection object =iterator.next(); - object.close(); - } - } + // checkRemotes(); + AtomicBoolean AB=new AtomicBoolean(true); try { - if (syn) { - KLALBBlock sbk = new KLALBBlock(); - sbk.cuid = tc.getCuid(); - sbk.number = 0; - sbk.command = 2; - sbk.lservice = tc.getServiceElement().name; - sbk.lipport = tc.getServiceElement().ipport; - klc.submitDataBlockNoDelay(sbk); - } - + //AtomicReferencetn=new AtomicReference<>(); + //AtomicReferenceltn=new AtomicReference<>(); Thread lt = ThreadTool.makeVThreadIfSupport("数据包计时重发线程", () -> { try { while (AB.get()) { klc.outputTimer(tc); + /*if(!checkRemotes()) { + AB.set(false); + }*/ } } catch (InterruptedException e) { - + e.printStackTrace(); + AB.set(false); + //tn.get().interrupt(); + //ltn.get().interrupt(); + tc.getOutputcache().clear(); + tc.close(); + KLALBBlock sbk = new KLALBBlock(); + sbk.cuid = tc.getCuid(); + sbk.number = 0; + sbk.command = 3; + klc.submitDataBlockNoDelay(sbk); + } catch (IOException e) { + e.printStackTrace(); } }); - Thread.sleep(50); + //Thread.sleep(500); Thread ls = ThreadTool.makeVThreadIfSupport("本地发送线程", () -> { try { - while (AB.get()) { + while (true) { KLALBBlock rd=klc.getDataBlock(tc); if(rd.data==null) { break; @@ -84,6 +86,7 @@ public class IOThreadManager { AB.set(false); lt.interrupt(); tc.getOutputcache().clear(); + tc.close(); KLALBBlock sbk = new KLALBBlock(); sbk.cuid = tc.getCuid(); sbk.number = 0; @@ -97,12 +100,14 @@ public class IOThreadManager { } } }); + //tn.set(ls); Thread lr = ThreadTool.makeVThreadIfSupport("本地接收线程", () -> { try { - while (AB.get()) { + while (true) { KLALBBlock ks = tc.packBlock(); klc.putDataBlock(tc, ks); if (ks.data == null) { + klc.waitOutput(tc); break; } } @@ -112,6 +117,7 @@ public class IOThreadManager { ls.interrupt(); lt.interrupt(); tc.getOutputcache().clear(); + tc.close(); KLALBBlock sbk = new KLALBBlock(); sbk.cuid = tc.getCuid(); sbk.number = 0; @@ -125,22 +131,34 @@ public class IOThreadManager { } } }); - + //ltn.set(lr); tc.setRSTHook((t)->{ + //System.out.println("xxxxxxxxxxxxxxxxxxxxxxxxxx"); + //System.exit(123); AB.set(false); ls.interrupt(); lt.interrupt(); tc.getOutputcache().clear(); + tc.close(); }); + if (syn) { + KLALBBlock sbk = new KLALBBlock(); + sbk.cuid = tc.getCuid(); + sbk.number = 0; + sbk.command = 2; + sbk.lservice = tc.getServiceElement().name; + sbk.lipport = tc.getServiceElement().ipport; + klc.submitAckBlockNoDelay(sbk,1); + } ls.start(); lr.start(); lt.start(); + //System.out.println(klc.getLocaltcps()); lr.join(); + ls.join(); ls.interrupt(); - - - klc.waitOutput(tc); lt.interrupt(); + } catch (InterruptedException e) { e.printStackTrace(); } finally { @@ -190,20 +208,30 @@ public class IOThreadManager { klc.submitDataBlock(tmp); } catch (InterruptedException e) { e.printStackTrace(); + } catch (IOException e) { + // TODO 自动生成的 catch 块 + e.printStackTrace(); } } - if(klc.getRemotetcps().isEmpty()) { - System.out.println("所有线路已断开，请检查网络"); - Collection c=klc.getLocaltcps().values(); - for (Iterator iterator = c.iterator(); iterator.hasNext();) { - LocalTCPConnection object =iterator.next(); - object.close(); - } - } + // checkRemotes(); } } + /*private boolean checkRemotes() { + boolean b=true; + if(klc.getRemotetcps().isEmpty()) { + System.out.println("所有线路已断开"); + b=false; + Collection c=klc.getLocaltcps().values(); + for (Iterator iterator = c.iterator(); iterator.hasNext();) { + LocalTCPConnection object =iterator.next(); + object.close(); + } + } + return b; + } +*/ public void close() { isopen = false; Collectionltcc= klc.getLocaltcps().values(); diff --git a/src/org/kne/cloud/network/klalb/KLALBBlock.java b/src/org/kne/cloud/network/klalb/KLALBBlock.java index 878867f..5cc04ad 100644 --- a/src/org/kne/cloud/network/klalb/KLALBBlock.java +++ b/src/org/kne/cloud/network/klalb/KLALBBlock.java @@ -14,7 +14,7 @@ public class KLALBBlock implements Comparable{ public volatile int cuid;//用于识别数据包的stream ID号 public volatile long number;//数据包的编号，用于排序 public volatile byte[]data;//数据内容 - public volatile int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST + public volatile int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST 4=KEEP public volatile long pingtime;//PING计时器 public volatile int cacheused; diff --git a/src/org/kne/cloud/network/klalb/KLALBCore.java b/src/org/kne/cloud/network/klalb/KLALBCore.java index 445633f..83da240 100644 --- a/src/org/kne/cloud/network/klalb/KLALBCore.java +++ b/src/org/kne/cloud/network/klalb/KLALBCore.java @@ -45,7 +45,7 @@ public class KLALBCore { private volatile int cacheblocks; - private Map localtcps = Collections.synchronizedMap(new HashMap<>()); + private Map localtcps = new ConcurrentHashMap<>(); private List remotetcps = new Vector<>(); private PredicateacceptSYN; @@ -194,8 +194,7 @@ public class KLALBCore { }else if(brc.command==1) { }else if(brc.command==2) { if(!localtcps.containsKey(brc.cuid)) { - localtcps.put(brc.cuid, null); - try { + if(acceptSYN==null||!acceptSYN.test(brc)) { KLALBBlock rst=new KLALBBlock(); rst.cuid=brc.cuid; @@ -203,27 +202,24 @@ public class KLALBCore { rst.command=3; submitDataBlockNoDelay(rst); } - }finally { - localtcps.remove(brc.cuid, null); - } } }else if(brc.command==3) { LocalTCPConnection ltc=localtcps.get(brc.cuid); + if(ltc!=null) { - ltc.close(); + //ltc.close(); ltc.getRSTHook().accept(ltc); } } }else if(brc.number>0){ LocalTCPConnection ltcs=localtcps.get(brc.cuid); if(ltcs!=null) { - ltcs.getSendDeque().add(brc); - KLALBBlock ack=new KLALBBlock(); - ack.cuid=brc.cuid; - ack.number=-brc.number; - ack.cacheused=ltcs.getSendDeque().size(); - submitAckBlockNoDelay(ack,3); + ack.cuid=brc.cuid; + ack.number=-brc.number; + ack.cacheused=ltcs.getSendDeque().size(); + submitAckBlockNoDelay(ack,3); + ltcs.getSendDeque().add(brc); } @@ -251,6 +247,12 @@ public class KLALBCore { return b.number == nx; });*/ ltc.setPeerCacheUsed(brc.cacheused); + }else { + KLALBBlock sbk = new KLALBBlock(); + sbk.cuid = brc.cuid; + sbk.number = 0; + sbk.command = 3; + submitDataBlockNoDelay(sbk); } int cuid=brc.cuid; synchronized (remotetcps) { @@ -318,12 +320,14 @@ public class KLALBCore { tc.setDelay((tc.getDelay()*10+delay)/11); } - public void submitDataBlock(KLALBBlock ks) throws InterruptedException { + public void submitDataBlock(KLALBBlock ks) throws InterruptedException, IOException { submitDataBlock(ks,1); } - public void submitDataBlock(KLALBBlock kb,int i) throws InterruptedException { + public void submitDataBlock(KLALBBlock kb,int i) throws InterruptedException, IOException { int ni=Math.min(i, remotetcps.size()); while(true) { + if(remotetcps.isEmpty()) + throw new IOException("发送错误"); synchronized (remotetcps) { sortRemoteTCPByPrediction(); for (Iterator iterator = remotetcps.iterator(); iterator.hasNext();) { @@ -374,8 +378,10 @@ public class KLALBCore { } } } - public void putDataBlock(LocalTCPConnection tc, KLALBBlock ks) throws InterruptedException { + public void putDataBlock(LocalTCPConnection tc, KLALBBlock ks) throws InterruptedException, IOException { while (true) { + if(remotetcps.isEmpty()) + throw new IOException("发送错误"); boolean b; synchronized (tc.getOutputcache()) { b=!tc.getOutputcache().isEmpty()&&(tc.getOutputcount()-tc.getOutputcache().get(0).number>cacheblocks); @@ -402,13 +408,15 @@ public class KLALBCore { }*/ submitDataBlock(ks); tc.getOutputcache().add(ks); - while( tc.getPeerCacheUsed()>5000) { + while( tc.getPeerCacheUsed()>2*cacheblocks) { + if(remotetcps.isEmpty()) + throw new IOException("发送错误"); Thread.sleep(1); } } - private volatile long uackt=System.nanoTime(); - public void outputTimer(LocalTCPConnection tc) throws InterruptedException { + //private volatile long uackt=System.nanoTime(); + public void outputTimer(LocalTCPConnection tc) throws InterruptedException, IOException { List l=tc.getOutputcache(); synchronized( l) { for (int i = 0; i < l.size(); i++) { @@ -434,28 +442,48 @@ public class KLALBCore { if(block.resend>=Consts.MAX_RESEND) { l.remove(i); i--; + throw new IOException("重传失败："+block); } } } } long curr=System.nanoTime(); - if(curr-uackt>Consts.UACK_TIME) { - uackt=curr; + if(tc.getSendDeque().size()Consts.UACK_TIME) { + tc.uackt=curr; KLALBBlock ack=new KLALBBlock(); ack.cuid=tc.getCuid(); ack.number=Long.MIN_VALUE; ack.cacheused=tc.getSendDeque().size(); submitAckBlockNoDelay(ack,3); } + /*if(tc.getSendDeque().size()Consts.KEEP_TIME) { + tc.ukeept=curr; + KLALBBlock keep=new KLALBBlock(); + keep.cuid=tc.getCuid(); + keep.number=0; + keep.command=4; + submitDataBlockNoDelay(keep); + }*/ Thread.sleep(10); } - public KLALBBlock getDataBlock(LocalTCPConnection tc) throws InterruptedException { - return tc.getSendDeque().take(); + public KLALBBlock getDataBlock(LocalTCPConnection tc) throws InterruptedException, IOException { + KLALBBlock recv=null; + while(true) { + if(remotetcps.isEmpty()) + throw new IOException("发送错误"); + recv=tc.getSendDeque().poll(); + if(recv!=null) + break; + Thread.sleep(1); + } + return recv; } - public void waitOutput(LocalTCPConnection tc) throws InterruptedException { + public void waitOutput(LocalTCPConnection tc) throws InterruptedException, IOException { while(tc.getOutputcache().size()>0) { + if(remotetcps.isEmpty()) + throw new IOException("发送错误"); Thread.sleep(10); } } diff --git a/src/org/kne/cloud/network/klalb/KLALBServer.java b/src/org/kne/cloud/network/klalb/KLALBServer.java index ec927f5..5d46989 100644 --- a/src/org/kne/cloud/network/klalb/KLALBServer.java +++ b/src/org/kne/cloud/network/klalb/KLALBServer.java @@ -22,7 +22,7 @@ import org.kne.cloud.network.mport.ThreadTool; public class KLALBServer { - Map whm=new WeakHashMap(); + Map whm=Collections.synchronizedMap( new WeakHashMap()); public KLALBServer(int port,Supplier gjso) throws IOException { TCPListener tcpl=new TCPListener(port); tcpl.setCon((s)->{ @@ -50,9 +50,34 @@ public class KLALBServer { tcc.handshake(10); - IOThreadManager nx = null; - synchronized (tcpl) { - if(whm.containsKey(uid)) { + IOThreadManager nx=whm.computeIfAbsent(uid, (kuid)->{ + IOThreadManager iom=new IOThreadManager(); + iom.getCore().setAcceptSYN((b)->{ + try { + STJson stj=new STJson(gjso.get()); + if(!stj.checkSafety(b.lipport)) { + System.out.println("未授权的请求:"+b.lipport); + return false; + } + Socket soc=new Socket(b.lipport.getIp(),b.lipport.getPort()); + LocalTCPConnection ltc=new LocalTCPConnection(soc, b.cuid); + ThreadTool.makeVThreadIfSupport("局域网连接监视线程", ()->{ + try { + iom.handleLocal(ltc, false); + }finally { + ltc.close(); + } + }).start(); + return true; + } catch (IOException e) { + System.out.println("连接本地服务"+b.lipport+"失败，请检查你的本地服务"); + e.printStackTrace(); + } + return false; + }); + return iom; + }); + /*if(whm.containsKey(uid)) { nx=whm.get(uid); }else { nx=new IOThreadManager(); @@ -81,8 +106,7 @@ public class KLALBServer { return false; }); whm.put(uid, nx); - } - } + }*/ nx.handleRemote(tcc); }catch(IOException e) { e.printStackTrace(); diff --git a/src/org/kne/cloud/network/klalb/LocalTCPConnection.java b/src/org/kne/cloud/network/klalb/LocalTCPConnection.java index bf57505..c94b746 100644 --- a/src/org/kne/cloud/network/klalb/LocalTCPConnection.java +++ b/src/org/kne/cloud/network/klalb/LocalTCPConnection.java @@ -25,10 +25,12 @@ private volatile long inputcount = 1; private volatile long outputcount = 1; public LocalTCPConnection(Socket s) throws IOException { super(s); + s.setKeepAlive(true); cuid=sng.getAndIncrement(); } public LocalTCPConnection(Socket s,int uid) throws IOException { super(s); + s.setKeepAlive(true); cuid=uid; } public int getCuid() { @@ -95,7 +97,10 @@ private volatile long outputcount = 1; return sendDeque; } private volatile int pcu=0; - private Consumer hook; + private volatile Consumer hook; + + public volatile long uackt=System.nanoTime(); + public volatile long ukeept=System.nanoTime(); public void setPeerCacheUsed(int cacheused) { pcu= cacheused; } diff --git a/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java b/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java index e9e5c8e..05d8195 100644 --- a/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java +++ b/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java @@ -331,6 +331,9 @@ public class RemoteTCPConnection extends TCPConnection { if(v==-1) { throw new EOFException(); } + if(v!=0) { + throw new StreamCorruptedException(); + } } } diff --git a/src/org/kne/cloud/network/klalb/TCPConnection.java b/src/org/kne/cloud/network/klalb/TCPConnection.java index 8c8ad99..7e92a4f 100644 --- a/src/org/kne/cloud/network/klalb/TCPConnection.java +++ b/src/org/kne/cloud/network/klalb/TCPConnection.java @@ -15,7 +15,7 @@ public class TCPConnection { - protected Socket connect; + protected volatile Socket connect; protected DataInputStream din; protected DataOutputStream dout; @@ -43,13 +43,15 @@ public class TCPConnection { e1.printStackTrace(); } try { - if (connect != null) + if (connect != null) { connect.close(); + } + } catch (IOException e) { // TODO 自动生成的 catch 块 e.printStackTrace(); } - connect = null; + //connect = null; } public TCPConnection( Socket soc) throws IOException { diff --git a/src/org/kne/ui/YScrollPane.java b/src/org/kne/ui/YScrollPane.java index b4e1310..a325511 100644 --- a/src/org/kne/ui/YScrollPane.java +++ b/src/org/kne/ui/YScrollPane.java @@ -19,6 +19,7 @@ import javax.swing.JScrollBar; public class YScrollPane extends JPanel { private JPanel settings; + private JScrollBar scrollBar; public YScrollPane(int contentx){ this(contentx,BorderLayout.EAST); } @@ -26,7 +27,7 @@ public class YScrollPane extends JPanel { setLayout(new BorderLayout(0, 0)); setBackground(Color.WHITE); - JScrollBar scrollBar = new JScrollBar(); + scrollBar = new JScrollBar(); add(scrollBar,d); JPanel panel = new JPanel(); @@ -125,5 +126,8 @@ settings.addContainerListener(new ContainerListener() { public JPanel getView() { return settings; } + public JScrollBar getScrollBar() { + return scrollBar; + } }