增加一些可配置的延迟参数，以便人工控制请求频率

codingmiao · codingmiao · commit 58564f6ddf61 · 2024-06-11T19:25:53.000+08:00
diff --git a/.gitignore b/.gitignore
@@ -4,10 +4,7 @@ target/
 !**/src/test/**/target/
 
 ### IntelliJ IDEA ###
-.idea/modules.xml
-.idea/jarRepositories.xml
-.idea/compiler.xml
-.idea/libraries/
+.idea/
 *.iws
 *.iml
 *.ipr
diff --git a/common/src/main/java/org/wowtools/hppt/common/client/ClientTalker.java b/common/src/main/java/org/wowtools/hppt/common/client/ClientTalker.java
@@ -90,10 +90,7 @@ public static byte[] buildSendToServerBytes(CommonConfig config, long maxSendBod
         }
 
         byte[] bytes = rBuilder.build().toByteArray();
-        //压缩 加密
-        if (config.enableCompress) {
-            bytes = BytesUtil.compress(bytes);
-        }
+        //加密
         if (config.enableEncrypt) {
             bytes = aesCipherUtil.encryptor.encrypt(bytes);
         }
@@ -111,13 +108,10 @@ public static boolean receiveServerBytes(CommonConfig config, byte[] responseBod
         }
         ProtoMessage.MessagePb rMessagePb;
         try {
-            //解密、解压
+            //解密
             if (config.enableEncrypt) {
                 responseBody = aesCipherUtil.descriptor.decrypt(responseBody);
             }
-            if (config.enableCompress) {
-                responseBody = BytesUtil.decompress(responseBody);
-            }
             log.debug("收到服务端发回字节数 {}", responseBody.length);
             rMessagePb = ProtoMessage.MessagePb.parseFrom(responseBody);
         } catch (Exception e) {
diff --git a/common/src/main/java/org/wowtools/hppt/common/server/ServerTalker.java b/common/src/main/java/org/wowtools/hppt/common/server/ServerTalker.java
@@ -26,14 +26,10 @@ public static void receiveClientBytes(CommonConfig config, ServerSessionManager
         if (null == bytes || bytes.length == 0) {
             return;
         }
-        //解密 解压
+        //解密
         if (config.enableEncrypt) {
             bytes = client.aesCipherUtil.descriptor.decrypt(bytes);
         }
-        if (config.enableCompress) {
-            bytes = BytesUtil.decompress(bytes);
-        }
-
         ProtoMessage.MessagePb inputMessage = ProtoMessage.MessagePb.parseFrom(bytes);
         Map<Integer, ServerSession> serverSessionMap = serverSessionManager.getServerSessionMapByClientId(client.clientId);
 
@@ -116,10 +112,7 @@ public static byte[] replyToClient(CommonConfig config, ServerSessionManager ser
 
 
         byte[] bytes = rBuilder.build().toByteArray();
-        //压缩 加密
-        if (config.enableCompress) {
-            bytes = BytesUtil.compress(bytes);
-        }
+        //加密
         if (config.enableEncrypt) {
             bytes = client.aesCipherUtil.encryptor.encrypt(bytes);
         }
diff --git a/common/src/main/java/org/wowtools/hppt/common/util/CommonConfig.java b/common/src/main/java/org/wowtools/hppt/common/util/CommonConfig.java
@@ -5,10 +5,10 @@
  * @date 2024/1/25
  */
 public class CommonConfig {
-    /**
-     * 是否启用压缩，默认启用
-     */
-    public boolean enableCompress = true;
+//    /**
+//     * 是否启用压缩，默认启用
+//     */
+//    public boolean enableCompress = true;
 
     /**
      * 是否启用内容加密，默认启用
diff --git a/run/src/main/java/org/wowtools/hppt/run/sc/common/ClientSessionService.java b/run/src/main/java/org/wowtools/hppt/run/sc/common/ClientSessionService.java
@@ -40,7 +40,7 @@ public abstract class ClientSessionService {
     private boolean firstLoginErr = true;
     private boolean noLogin = true;
 
-    protected volatile boolean actived = true;
+    protected volatile boolean running = true;
 
     /**
      * 当一个事件结束时发起的回调
@@ -145,7 +145,7 @@ protected void receiveServerBytes(byte[] bytes) throws Exception {
      * 当发生难以修复的异常等情况时，主动调用此方法结束当前服务，以便后续自动重启等操作
      */
     public void exit() {
-        actived = false;
+        running = false;
         clientSessionManager.close();
         try {
             doClose();
@@ -180,7 +180,7 @@ public void sync() {
 
     private Thread buildSendThread() {
         return new Thread(() -> {
-            while (actived) {
+            while (running) {
                 try {
                     byte[] sendBytes = ClientTalker.buildSendToServerBytes(config, config.maxSendBodySize, sendCommandQueue, sendBytesQueue, aesCipherUtil, true);
                     if (null != sendBytes) {
diff --git a/run/src/main/java/org/wowtools/hppt/run/sc/post/PostClientSessionService.java b/run/src/main/java/org/wowtools/hppt/run/sc/post/PostClientSessionService.java
@@ -8,6 +8,7 @@
 import org.wowtools.hppt.run.sc.common.ClientSessionService;
 import org.wowtools.hppt.run.sc.pojo.ScConfig;
 
+import java.nio.charset.StandardCharsets;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.UUID;
@@ -45,32 +46,32 @@ protected void connectToServer(ScConfig config, Cb cb) {
     private void startSendThread(Cb cb) {
         Thread.startVirtualThread(() -> {
             //等待初始化完成
-            while (null == sendQueue) {
-                try {
-                    Thread.sleep(10);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
-            }
             cb.end();
+            //起一个while循环不断发送数据
             final long sendSleepTime = config.post.sendSleepTime;
-            while (actived) {
-                if (sendSleepTime > 0) {
-                    try {
-                        Thread.sleep(sendSleepTime);
-                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
-                    }
-                }
+            while (running) {
                 try {
                     byte[] sendBytes;
                     List<byte[]> bytesList = new LinkedList<>();
                     sendBytes = sendQueue.take();
                     bytesList.add(sendBytes);
+                    if (sendSleepTime > 0) {
+                        try {
+                            Thread.sleep(sendSleepTime);
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException(e);
+                        }
+                    }
                     sendQueue.drainTo(bytesList);
                     sendBytes = BytesUtil.bytesCollection2PbBytes(bytesList);
-                    try (Response ignored = HttpUtil.doPost(sendUrl, sendBytes)) {
-                        log.debug("startSendThread 发送完成");
+                    try (Response r = HttpUtil.doPost(sendUrl, sendBytes)) {
+                        assert r.body() != null;
+                        byte[] rBytes = r.body().bytes();
+                        if (rBytes.length == 0) {
+                            log.debug("SendThread 发送完成");
+                        } else {
+                            throw new RuntimeException("异常的响应值" + new String(rBytes, StandardCharsets.UTF_8));
+                        }
                     }
                 } catch (Exception e) {
                     log.warn("SendThread异常", e);
@@ -86,24 +87,23 @@ private void startReplyThread() {
         Thread.startVirtualThread(() -> {
             boolean empty = false;
             final long sendSleepTime = config.post.sendSleepTime;
-            while (actived) {
-                if (sendSleepTime > 0) {
-                    try {
-                        Thread.sleep(sendSleepTime);
-                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
-                    }
-                }
+            while (running) {
+                //检测是否需要挂起接收线程
                 if (empty && !isNoLogin() && clientSessionManager.getSessionNum() == 0) {
                     synchronized (replyThreadEmptyLock) {
-                        log.info("无客户端,睡眠接收线程");
+                        log.info("无客户端,挂起接收线程");
                         try {
                             replyThreadEmptyLock.wait();
                         } catch (InterruptedException e) {
                             log.info("唤醒接收线程");
+                            if (!running) {
+                                log.info("退出已停止ReplyThread");
+                                return;
+                            }
                         }
                     }
                 }
+                //发一个接收请求接数据
                 try {
                     byte[] responseBytes;
                     try (Response response = HttpUtil.doPost(replyUrl, null)) {
@@ -112,18 +112,32 @@ private void startReplyThread() {
                     }
                     if (null != responseBytes && responseBytes.length > 0) {
                         log.debug("收到服务端响应字节数 {}", responseBytes.length);
-                        List<byte[]> bytesList = BytesUtil.pbBytes2BytesList(responseBytes);
+                        List<byte[]> bytesList;
+                        try {
+                            bytesList = BytesUtil.pbBytes2BytesList(responseBytes);
+                        } catch (Exception e) {
+                            log.warn("解析protobuf字节异常 {}", new String(responseBytes, StandardCharsets.UTF_8));
+                            throw e;
+                        }
                         for (byte[] bytes : bytesList) {
                             receiveServerBytes(bytes);
                         }
                         empty = bytesList.isEmpty();
-                    }else {
+                    } else {
                         empty = true;
                     }
                 } catch (Exception e) {
                     log.warn("ReplyThread异常", e);
                     exit();
                 }
+                //按需做等待
+                if (sendSleepTime > 0) {
+                    try {
+                        Thread.sleep(sendSleepTime);
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
             }
         });
     }
@@ -140,4 +154,14 @@ protected void newConnected() {
             replyThreadEmptyLock.notify();
         }
     }
+
+    @Override
+    public void exit() {
+        super.exit();
+        Thread.startVirtualThread(() -> {
+            synchronized (replyThreadEmptyLock) {
+                replyThreadEmptyLock.notify();
+            }
+        });
+    }
 }
diff --git a/run/src/main/java/org/wowtools/hppt/run/ss/common/ServerSessionService.java b/run/src/main/java/org/wowtools/hppt/run/ss/common/ServerSessionService.java
@@ -80,7 +80,6 @@ public void receiveClientBytes(CTX ctx, byte[] bytes) {
         if (null == bytes || bytes.length == 0) {
             return;
         }
-        //TODO 锁ctx异步
         log.debug("收到客户端字节数 {}", bytes.length);
         // 若客户端为空,则进行对时或登录
         ClientCell clientCell = ctxClientCellMap.get(ctx);
diff --git a/run/src/main/java/org/wowtools/hppt/run/ss/pojo/SsConfig.java b/run/src/main/java/org/wowtools/hppt/run/ss/pojo/SsConfig.java
@@ -45,9 +45,14 @@ public class SsConfig extends CommonConfig {
 
     public static final class PostConfig {
         /**
-         * 等待真实端口返回数据的毫秒数
+         * 等待真实端口返回数据的毫秒数，一般设一个略小于http服务超时时间的值
          */
         public long waitResponseTime = 30000;
+
+        /**
+         * 回复的servlet人为设置的延迟，避免客户端过于频繁的发请求
+         */
+        public long replyDelayTime = 0;
     }
 
     public PostConfig post = new PostConfig();
diff --git a/run/src/main/java/org/wowtools/hppt/run/ss/post/PostCtx.java b/run/src/main/java/org/wowtools/hppt/run/ss/post/PostCtx.java
@@ -15,6 +15,4 @@ public PostCtx(String cookie) {
         this.cookie = cookie;
     }
 
-    Thread waitResponseThread;
-
 }
diff --git a/run/src/main/java/org/wowtools/hppt/run/ss/post/PostServerSessionService.java b/run/src/main/java/org/wowtools/hppt/run/ss/post/PostServerSessionService.java
@@ -5,6 +5,7 @@
 import jakarta.servlet.http.HttpServletResponse;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.servlet.ErrorPageErrorHandler;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
@@ -32,10 +33,18 @@ public PostServerSessionService(SsConfig ssConfig) throws Exception {
     @Override
     public void init(SsConfig ssConfig) throws Exception {
         log.info("*********");
-        server = new Server(ssConfig.port);
+        server = new Server();
+        // 创建一个ServerConnector
+        ServerConnector connector = new ServerConnector(server);
+        connector.setPort(ssConfig.port);
+        // 设置请求超时时间（以毫秒为单位）
+        connector.setIdleTimeout(ssConfig.post.waitResponseTime * 2);
+        // 将connector添加到server
+        server.addConnector(connector);
+
         ServletContextHandler context = new ServletContextHandler(server, "/");
         context.addServlet(new ServletHolder(new SendServlet(this)), "/s");
-        context.addServlet(new ServletHolder(new ReplyServlet(this, ssConfig.post.waitResponseTime)), "/r");
+        context.addServlet(new ServletHolder(new ReplyServlet(this, ssConfig.post.waitResponseTime, ssConfig.post.replyDelayTime)), "/r");
         context.addServlet(new ServletHolder(new ErrorServlet()), "/e");
 
         ErrorPageErrorHandler errorHandler = new ErrorPageErrorHandler();
diff --git a/run/src/main/java/org/wowtools/hppt/run/ss/post/ReplyServlet.java b/run/src/main/java/org/wowtools/hppt/run/ss/post/ReplyServlet.java
@@ -1,6 +1,5 @@
 package org.wowtools.hppt.run.ss.post;
 
-import jakarta.servlet.ServletException;
 import jakarta.servlet.http.HttpServlet;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
@@ -21,13 +20,14 @@
  */
 @Slf4j
 final class ReplyServlet extends HttpServlet {
-    private static final byte[] emptyBytes = new byte[0];
     private final PostServerSessionService postServerSessionService;
     private final long waitResponseTime;
+    private final long replyDelayTime;
 
-    public ReplyServlet(PostServerSessionService postServerSessionService, long waitResponseTime) {
+    public ReplyServlet(PostServerSessionService postServerSessionService, long waitResponseTime,long replyDelayTime) {
         this.postServerSessionService = postServerSessionService;
         this.waitResponseTime = waitResponseTime;
+        this.replyDelayTime = replyDelayTime;
     }
 
 
@@ -43,19 +43,29 @@ protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws I
 
     //取缓冲区中的数据返回
     private void write(PostCtx ctx, HttpServletResponse resp) throws IOException {
+        if (replyDelayTime > 0) {
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
         List<byte[]> bytesList = new LinkedList<>();
         byte[] rBytes;
         try {
             rBytes = ctx.sendQueue.poll(waitResponseTime, TimeUnit.MILLISECONDS);
         } catch (InterruptedException e) {
             rBytes = null;
         }
-        if (null != rBytes) {
-            bytesList.add(rBytes);
-            ctx.sendQueue.drainTo(bytesList);
+        if (null == rBytes) {
+            return;
         }
+        bytesList.add(rBytes);
+
+        ctx.sendQueue.drainTo(bytesList);
         rBytes = BytesUtil.bytesCollection2PbBytes(bytesList);
-        log.debug("向客户端发送字节 body {}", rBytes.length);
+        log.debug("向客户端发送字节 bytesList {} body {}", bytesList.size(), rBytes.length);
         try (OutputStream os = resp.getOutputStream()) {
             os.write(rBytes);
         }
diff --git a/run/src/main/java/org/wowtools/hppt/run/ss/post/WaitServlet.java b/run/src/main/java/org/wowtools/hppt/run/ss/post/WaitServlet.java

Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,6 @@ public void receiveClientBytes(CTX ctx, byte[] bytes) {`
`80`	`80`	`if (null == bytes \|\| bytes.length == 0) {`
`81`	`81`	`return;`
`82`	`82`	`}`
`83`		`- //TODO 锁ctx异步`
`84`	`83`	`log.debug("收到客户端字节数 {}", bytes.length);`
`85`	`84`	`// 若客户端为空,则进行对时或登录`
`86`	`85`	`ClientCell clientCell = ctxClientCellMap.get(ctx);`
Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,4 @@ public PostCtx(String cookie) {`
`15`	`15`	`this.cookie = cookie;`
`16`	`16`	`}`
`17`	`17`
`18`		`- Thread waitResponseThread;`
`19`		`-`
`20`	`18`	`}`