refactor(core):优化Redis会话处理逻辑

- 引入线程持有机制，确保数据写入的线程安全性- 移除冗余的异常打印代码，统一异常处理方式- 优化空字符串处理，复用空输出流实例- 调整客户端连接的获取与释放时机，提高资源利用率
- 添加线程安全检查，防止跨线程操作缓冲区

Author: smthing

src/main/java/tech/smartboot/redisun/RedisMessageProcessor.java

@@ -126,7 +126,7 @@ public void stateEvent0(AioSession session, StateMachineEnum stateMachineEnum, T
 
         // 如果有异常发生，打印异常堆栈信息
         if (throwable != null) {
-            throwable.printStackTrace();
+//            throwable.printStackTrace();
         }
     }
 }

src/main/java/tech/smartboot/redisun/RedisSession.java

@@ -34,6 +34,15 @@ final class RedisSession {
      */
     private RESP decodingResponse;
 
+    private Thread holdThread;
+
+    public Thread getHoldThread() {
+        return holdThread;
+    }
+
+    public void setHoldThread(Thread holdThread) {
+        this.holdThread = holdThread;
+    }
 
     /**
      * 获取正在解码的响应对象

src/main/java/tech/smartboot/redisun/Redisun.java

@@ -510,10 +510,10 @@ private CompletableFuture<RESP> execute(Command command) {
         // 创建用于接收结果的CompletableFuture
         CompletableFuture<RESP> future = new CompletableFuture<>();
         // 设置异常处理
-        future.exceptionally(t -> {
-            t.printStackTrace();
-            return null;
-        });
+//        future.exceptionally(t -> {
+//            t.printStackTrace();
+//            return null;
+//        });
         AioQuickClient client = null;
         try {
             // 构建命令参数
@@ -524,17 +524,21 @@ private CompletableFuture<RESP> execute(Command command) {
             // 获取可用的客户端连接
             client = multiplexClient.acquire();
             AioSession session = client.getSession();
-
+            RedisSession redisSession = session.getAttachment();
+            redisSession.setHoldThread(Thread.currentThread());
+            multiplexClient.reuse(client);
             synchronized (client) {
-                RedisSession redisSession = session.getAttachment();
+
                 // 设置当前命令的future
                 redisSession.offer(future);
                 // 将命令编码为RESP格式并写入缓冲区
                 arrays.writeTo(session.writeBuffer());
             }
-            multiplexClient.reuse(client);
+
             // 刷新缓冲区，发送数据
-            session.writeBuffer().flush();
+            if (redisSession.getHoldThread() == Thread.currentThread()) {
+                session.writeBuffer().flush();
+            }
         } catch (Throwable e) {
             // 发生异常时完成future
             if (client != null) {

src/main/java/tech/smartboot/redisun/resp/BulkStrings.java

@@ -43,6 +43,7 @@ public class BulkStrings extends RESP<String> {
 
     // 用于存储字符串数据的输出流
     private ByteArrayOutputStream out;
+    private final static ByteArrayOutputStream EMPTY_OUTPUT_STREAM = new ByteArrayOutputStream(0);
 
     /**
      * 私有构造函数，防止外部直接实例化
@@ -73,7 +74,7 @@ public boolean decode(ByteBuffer readBuffer) {
                     if (length == 0) {
                         // 空字符串
                         state = DECODE_STATE_END;
-                        out = new ByteArrayOutputStream(0);
+                        out = EMPTY_OUTPUT_STREAM;
                     } else if (length > 0) {
                         // 初始化输出流以存储字符串数据
                         out = new ByteArrayOutputStream(length);