feat(redis):优化Redis会话命令写入机制- 引入WriteBuffer依赖以支持高效数据写入

-重构RedisSession构造函数，接受WriteBuffer参数
- 新增writeCommand方法处理命令写入和并发控制
- 实现基于信号量的命令队列处理机制
- 添加Tuple内部类用于存储命令和Future的组合
- 移除旧的计数器逻辑，简化会话管理流程- 在RedisMessageProcessor中调用flush确保数据及时发送

Author: smthing

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

@@ -96,6 +96,7 @@ public void process0(AioSession session, RESP msg) {
         } else {
             future.complete(msg);
         }
+        redisSession.flush();
     }
 
     /**
@@ -114,7 +115,7 @@ public void stateEvent0(AioSession session, StateMachineEnum stateMachineEnum, T
             // 处理新建会话事件
             case NEW_SESSION: {
                 // 为新会话创建并绑定Redis会话对象
-                RedisSession redisSession = new RedisSession();
+                RedisSession redisSession = new RedisSession(session.writeBuffer());
                 session.setAttachment(redisSession);
             }
             break;

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

@@ -1,9 +1,12 @@
 package tech.smartboot.redisun;
 
+import org.smartboot.socket.transport.WriteBuffer;
 import tech.smartboot.redisun.resp.RESP;
 
+import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Semaphore;
 
 /**
  * Redis会话管理类
@@ -36,14 +39,14 @@ final class RedisSession {
 
     private int offerCount = 0;
     private int pollCount = 0;
+    private final WriteBuffer writeBuffer;
 
-    public int incrOfferCount() {
-        return ++offerCount;
+    public RedisSession(WriteBuffer writeBuffer) {
+        this.writeBuffer = writeBuffer;
     }
 
-    public int getOfferCount() {
-        return offerCount;
-    }
+    private final ConcurrentLinkedQueue<Tuple> commandQueue = new ConcurrentLinkedQueue<>();
+    private final Semaphore semaphore = new Semaphore(1);
 
     /**
      * 获取正在解码的响应对象
@@ -68,12 +71,43 @@ public CompletableFuture<RESP> poll() {
         return pipeline.poll();
     }
 
-    public void offer(CompletableFuture<RESP> future) {
-        pipeline.offer(future);
+    public void writeCommand(CompletableFuture<RESP> future, Command command) throws IOException {
+        if (semaphore.tryAcquire()) {
+            offerCount++;
+            pipeline.offer(future);
+            command.writeTo(writeBuffer);
+            Tuple tuple;
+            while ((tuple = commandQueue.poll()) != null) {
+                pipeline.offer(tuple.future);
+                tuple.command.writeTo(writeBuffer);
+            }
+            writeBuffer.flush();
+            semaphore.release();
+//            flush();
+        } else {
+            offerCount++;
+            commandQueue.offer(new Tuple(future, command));
+        }
     }
 
-    public int getPollCount() {
-        return pollCount;
+    public void flush() {
+        if (commandQueue.isEmpty() || !semaphore.tryAcquire()) {
+            return;
+        }
+
+        try {
+            Tuple tuple;
+            while ((tuple = commandQueue.poll()) != null) {
+                offerCount++;
+                pipeline.offer(tuple.future);
+                tuple.command.writeTo(writeBuffer);
+            }
+        } catch (Throwable e) {
+            throw new RedisunException(e);
+        } finally {
+            semaphore.release();
+        }
+
     }
 
     int load() {
@@ -82,4 +116,13 @@ int load() {
         return size >= 0 ? size : -size;
     }
 
+    private static class Tuple {
+        private final CompletableFuture<RESP> future;
+        private final Command command;
+
+        public Tuple(CompletableFuture<RESP> future, Command command) {
+            this.future = future;
+            this.command = command;
+        }
+    }
 }

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

@@ -541,20 +541,7 @@ private CompletableFuture<RESP> execute(Command command) {
                     future.thenRun(() -> multiplexClient.reuse(finalClient));
                 }
             }
-
-            int offerCount = redisSession.incrOfferCount();
-            int pollCount = redisSession.getPollCount();
-
-            synchronized (client) {
-                // 设置当前命令的future
-                redisSession.offer(future);
-                command.writeTo(session.writeBuffer());
-            }
-
-            // 刷新缓冲区，发送数据
-            if (offerCount == redisSession.getOfferCount() && pollCount == redisSession.getPollCount()) {
-                session.writeBuffer().flush();
-            }
+            redisSession.writeCommand(future, command);
         } catch (Throwable e) {
             // 发生异常时完成future
             if (client != null) {