增加一个BufferPool对象，统一管理阻塞队列

Author: codingmiao

run/src/main/java/org/wowtools/hppt/common/client/ClientTalker.java

@@ -3,10 +3,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.wowtools.hppt.common.pojo.SessionBytes;
 import org.wowtools.hppt.common.pojo.TalkMessage;
-import org.wowtools.hppt.common.util.AesCipherUtil;
-import org.wowtools.hppt.common.util.CommonConfig;
-import org.wowtools.hppt.common.util.Constant;
-import org.wowtools.hppt.common.util.DebugConfig;
+import org.wowtools.hppt.common.util.*;
 
 import java.nio.charset.StandardCharsets;
 import java.util.LinkedList;
@@ -35,7 +32,7 @@ public class ClientTalker {
      * @throws Exception
      */
     public static byte[] buildSendToServerBytes(CommonConfig config, long maxSendBodySize,
-                                                BlockingQueue<String> sendCommandQueue, BlockingQueue<SessionBytes> sendBytesQueue,
+                                                BufferPool<String> sendCommandQueue, BufferPool<SessionBytes> sendBytesQueue,
                                                 AesCipherUtil aesCipherUtil, boolean wait) throws Exception {
         long sendBodySize = 0;//大致预估发送体积
         //命令
@@ -99,7 +96,7 @@ public static byte[] buildSendToServerBytes(CommonConfig config, long maxSendBod
 
     //接收服务端发来的字节并做相应处理
     public static boolean receiveServerBytes(CommonConfig config, byte[] responseBody,
-                                             ClientSessionManager clientSessionManager, AesCipherUtil aesCipherUtil, BlockingQueue<String> sendCommandQueue,
+                                             ClientSessionManager clientSessionManager, AesCipherUtil aesCipherUtil, BufferPool<String> sendCommandQueue,
                                              Map<Integer, ClientBytesSender.SessionIdCallBack> sessionIdCallBackMap) throws Exception {
         if (null == responseBody) {
             return true;

run/src/main/java/org/wowtools/hppt/common/util/BufferPool.java

@@ -0,0 +1,87 @@
+package org.wowtools.hppt.common.util;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 缓冲池，内置一个LinkedBlockingQueue,用以解耦生产者和消费者、缓冲数据并做监控
+ *
+ * @author liuyu
+ * @date 2024/10/27
+ */
+public class BufferPool<T> {
+    private final LinkedBlockingQueue<T> queue = new LinkedBlockingQueue<>();
+
+    private final String name;
+
+    /**
+     *
+     * @param name 缓冲池名字，为便于排查，请保证业务功能上的唯一
+     */
+    public BufferPool(String name) {
+        this.name = name;
+    }
+
+    /**
+     * 添加
+     *
+     * @param t t
+     */
+    public void add(T t) {
+        queue.add(t);
+    }
+
+    /**
+     * 获取,队列为空则一直阻塞等待
+     *
+     * @return t
+     */
+    public T take() {
+        try {
+            return queue.take();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * 获取,队列为空则返回null
+     *
+     * @return t or null
+     */
+    public T poll() {
+        return queue.poll();
+    }
+
+    /**
+     * 获取,队列为空则阻塞等待一段时间,超时则返回null
+     *
+     * @param timeout timeout
+     * @param unit    TimeUnit
+     * @return t or null
+     */
+    public T poll(long timeout, TimeUnit unit) {
+        T t;
+        try {
+            t = queue.poll(timeout, unit);
+        } catch (InterruptedException e) {
+            return null;
+        }
+        return t;
+    }
+
+    /**
+     * 获取队列中当前可用的所有元素,队列为空则阻塞等待，所以list至少会有一个元素
+     *
+     * @return list
+     */
+    public List<T> takeAndDrainToList() {
+        List<T> list = new LinkedList<>();
+        T t0 = take();
+        list.add(t0);
+        queue.drainTo(list);
+        return list;
+    }
+}

run/src/main/java/org/wowtools/hppt/run/sc/common/PortReceiver.java

@@ -15,9 +15,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -31,8 +29,8 @@ final class PortReceiver implements Receiver {
     private final ClientSessionService clientSessionService;
 
 
-    private final BlockingQueue<String> sendCommandQueue = new LinkedBlockingQueue<>();
-    private final BlockingQueue<SessionBytes> sendBytesQueue = new LinkedBlockingQueue<>();
+    private final BufferPool<String> sendCommandQueue = new BufferPool<>(">PortReceiver-sendCommand");
+    private final BufferPool<SessionBytes> sendBytesQueue = new BufferPool<>(">PortReceiver-sendBytesQueue");
 
     private final Map<Integer, ClientBytesSender.SessionIdCallBack> sessionIdCallBackMap = new ConcurrentHashMap<>();//<newSessionFlag,cb>
     private AesCipherUtil aesCipherUtil;

run/src/main/java/org/wowtools/hppt/run/sc/common/SsReceiver.java

@@ -1,18 +1,17 @@
 package org.wowtools.hppt.run.sc.common;
 
 import org.wowtools.hppt.common.client.ClientSession;
+import org.wowtools.hppt.common.util.BufferPool;
 import org.wowtools.hppt.common.util.RoughTimeUtil;
 import org.wowtools.hppt.run.sc.pojo.ScConfig;
 
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * @author liuyu
  * @date 2024/9/27
  */
 public final class SsReceiver implements Receiver {
-    public final BlockingQueue<byte[]> serverBytesQueue = new LinkedBlockingQueue<>();
+    public final BufferPool<byte[]> serverBytesQueue = new BufferPool<>(">SsReceiver-serverBytesQueue");
 
     private long lastUsedTime = RoughTimeUtil.getTimestamp();
 

run/src/main/resources/debug.ini

@@ -2,6 +2,6 @@
 
 
 # netty内存泄露检查级别 0 DISABLED 1 SIMPLE 2 ADVANCED 3 PARANOID
-NettyResourceLeakDetectorLevel = 0
+NettyResourceLeakDetectorLevel = 3
 # 是否开启消息流水号 1为开启 用于调试消息后发先至等问题，非调试时流水号为空
-OpenSerialNumber = 0
+OpenSerialNumber = 1

run/src/main/resources/logback.xml

@@ -27,7 +27,7 @@
     </appender>
 
     <!-- Root logger configuration -->
-    <root level="INFO">
+    <root level="DEBUG">
 <!--        <appender-ref ref="FILE" />-->
         <appender-ref ref="CONSOLE" />
     </root>