refactor(session):重构Redis会话命令写入逻辑

- 移除RedisSession中的WriteBuffer依赖
- 简化命令队列处理流程
- 移除信号量同步机制
-优化数据刷写逻辑
- 移除内部Tuple类定义
- 调整会话创建初始化参数

Author: smthing

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

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

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

@@ -1,12 +1,9 @@
 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会话管理类
@@ -39,14 +36,14 @@ final class RedisSession {
 
     private int offerCount = 0;
     private int pollCount = 0;
-    private final WriteBuffer writeBuffer;
 
-    public RedisSession(WriteBuffer writeBuffer) {
-        this.writeBuffer = writeBuffer;
+    public int incrOfferCount() {
+        return ++offerCount;
     }
 
-    private final ConcurrentLinkedQueue<Tuple> commandQueue = new ConcurrentLinkedQueue<>();
-    private final Semaphore semaphore = new Semaphore(1);
+    public int getOfferCount() {
+        return offerCount;
+    }
 
     /**
      * 获取正在解码的响应对象
@@ -71,41 +68,12 @@ public CompletableFuture<RESP> poll() {
         return pipeline.poll();
     }
 
-    public void writeCommand(CompletableFuture<RESP> future, Command command) throws IOException {
-        offerCount++;
-        if (semaphore.tryAcquire()) {
-            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 {
-            commandQueue.offer(new Tuple(future, command));
-        }
+    public void offer(CompletableFuture<RESP> future) {
+        pipeline.offer(future);
     }
 
-    public void flush() {
-        if (commandQueue.isEmpty() || !semaphore.tryAcquire()) {
-            return;
-        }
-
-        try {
-            Tuple tuple;
-            while ((tuple = commandQueue.poll()) != null) {
-                pipeline.offer(tuple.future);
-                tuple.command.writeTo(writeBuffer);
-            }
-        } catch (Throwable e) {
-            throw new RedisunException(e);
-        } finally {
-            semaphore.release();
-        }
-
+    public int getPollCount() {
+        return pollCount;
     }
 
     int load() {
@@ -114,13 +82,4 @@ 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,7 +541,20 @@ private CompletableFuture<RESP> execute(Command command) {
                     future.thenRun(() -> multiplexClient.reuse(finalClient));
                 }
             }
-            redisSession.writeCommand(future, command);
+
+            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();
+            }
         } catch (Throwable e) {
             // 发生异常时完成future
             if (client != null) {