ProcessWebSocketMessagesV2 版本

src/X1.WebSocket/Middleware/WebSocketMiddleware.cs

@@ -73,7 +73,6 @@ public class WebSocketMiddleware
     private readonly IWebSocketMessageQueueManager _messageQueueManager;
     private readonly ILogger<WebSocketMiddleware> _logger;
     private readonly WebSocketOptions _options;
-    private readonly WebSocketMessageBuffer _messageBuffer;
     private readonly WebSocketErrorHandler _errorHandler;
     private readonly WebSocketPerformanceMonitor _performanceMonitor;
     // 使用 WebSocket Channel 管理器来管理 Channel 的生命周期
@@ -100,7 +99,6 @@ public class WebSocketMiddleware
         _messageQueueManager = messageQueueManager;
         _logger = logger;
         _options = options.Value;
-        _messageBuffer = new WebSocketMessageBuffer(_options.MaxMessageSize);
         _errorHandler = new WebSocketErrorHandler(logger, _options.MessageRetryCount, _options.MessageRetryInterval);
         _performanceMonitor = new WebSocketPerformanceMonitor(logger);
         
@@ -169,8 +167,8 @@ public class WebSocketMiddleware
         
         // 使用原子操作的状态标志，确保线程安全
         var processingState = new AtomicBoolean(false);
-        // 从内存池租借缓冲区，用于接收消息
-        var buffer = ArrayPool<byte>.Shared.Rent(1024 * 4);
+        // 从内存池租借缓冲区，用于接收消息，大小与最大消息大小一致
+        var buffer = ArrayPool<byte>.Shared.Rent(_options.MaxMessageSize);
         // 使用 Channel 管理器创建新的消息通道
         var messageChannel = await _channelManager.CreateNewMessageChannel(connectionId);
         
@@ -308,8 +306,10 @@ public class WebSocketMiddleware
     {
         _logger.LogInformation("开始处理 WebSocket 消息，连接ID：{ConnectionId}", connectionId);
         
-        // 为每个连接创建独立的消息缓冲区
-        using var messageBuffer = new WebSocketMessageBuffer(_options.MaxMessageSize);
+        // 为每个连接创建独立的消息缓冲区，使用更大的缓冲区大小来容纳分片消息
+        // 缓冲区大小设置为 MaxMessageSize 的 1.5 倍，确保有足够空间处理分片
+        var bufferSize = (int)(_options.MaxMessageSize * 1.5);
+        using var messageBuffer = new WebSocketMessageBuffer(bufferSize);
         var lastMessageTime = DateTime.UtcNow;
         
         try
@@ -387,7 +387,7 @@ public class WebSocketMiddleware
         {
             _logger.LogError(ex, "WebSocket 消息处理发生未知异常，连接ID：{ConnectionId}，错误：{Error}", 
                 connectionId, ex.Message);
-            throw;
+            await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
         }
     }
 
@@ -419,25 +419,38 @@ public class WebSocketMiddleware
     {
         var processingStartTime = DateTime.UtcNow;
 
-        var success = await _errorHandler.HandleWithRetryAsync(async () =>
+        try
         {
-            // 检查缓冲区是否已满
+            // 检查当前缓冲区大小加上新数据是否会超过限制
+            var totalSize = messageBuffer.Size + receiveResult.Count;
+            if (totalSize > _options.MaxMessageSize)
+            {
+                _logger.LogWarning("消息大小将超过限制，连接ID：{ConnectionId}，当前缓冲区大小：{CurrentSize}，新数据大小：{NewDataSize}，总大小：{TotalSize}，限制：{MaxSize}", 
+                    connectionId, messageBuffer.Size, receiveResult.Count, totalSize, _options.MaxMessageSize);
+                throw new WebSocketException("消息大小超过限制");
+            }
+
+            // 写入数据到缓冲区（由于已经检查了大小，这里应该总是成功）
             if (!messageBuffer.TryWrite(buffer, 0, receiveResult.Count))
             {
-                _logger.LogWarning("消息缓冲区已满，连接ID：{ConnectionId}，缓冲区大小：{BufferSize}，消息大小：{MessageSize}", 
+                _logger.LogError("消息缓冲区写入失败，连接ID：{ConnectionId}，缓冲区大小：{BufferSize}，消息大小：{MessageSize}，这不应该发生", 
                     connectionId, messageBuffer.Size, receiveResult.Count);
-                throw new WebSocketException("消息缓冲区溢出");
+                throw new WebSocketException("消息缓冲区写入失败");
             }
 
+            _logger.LogDebug("数据已写入缓冲区，连接ID：{ConnectionId}，当前缓冲区大小：{BufferSize}，消息是否完整：{IsComplete}", 
+                connectionId, messageBuffer.Size, receiveResult.EndOfMessage);
+
             // 检查消息是否完整
             if (receiveResult.EndOfMessage)
             {
                 var messageData = messageBuffer.GetMessage();
                 
-                // 检查消息大小是否超过限制
+                // 由于已经在上面的 totalSize 检查中验证了大小，这里不需要再次检查
+                // 但为了安全起见，保留这个检查作为最后的验证
                 if (messageData.Length > _options.MaxMessageSize)
                 {
-                    _logger.LogWarning("消息大小超过限制，连接ID：{ConnectionId}，消息大小：{MessageSize}，限制：{MaxSize}", 
+                    _logger.LogError("最终消息大小超过限制，连接ID：{ConnectionId}，消息大小：{MessageSize}，限制：{MaxSize}，这不应该发生", 
                         connectionId, messageData.Length, _options.MaxMessageSize);
                     throw new WebSocketException("消息大小超过限制");
                 }
@@ -472,14 +485,39 @@ public class WebSocketMiddleware
                     throw new InvalidOperationException("消息通道已关闭");
                 }
 
-                // 重置缓冲区
+                // 重置缓冲区，准备接收下一条消息
                 messageBuffer.Reset();
+                _logger.LogDebug("缓冲区已重置，连接ID：{ConnectionId}", connectionId);
             }
-        });
-
-        if (!success)
+            else
+            {
+                _logger.LogDebug("消息分片已累积，连接ID：{ConnectionId}，当前缓冲区大小：{BufferSize}字节", 
+                    connectionId, messageBuffer.Size);
+            }
+        }
+        catch (WebSocketException ex)
         {
-            await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
+            // 对于 WebSocket 异常，直接抛出，不进行重试
+            _logger.LogError(ex, "WebSocket 消息处理异常，连接ID：{ConnectionId}，错误：{Error}", 
+                connectionId, ex.Message);
+            throw;
+        }
+        catch (Exception ex)
+        {
+            // 对于其他异常，使用重试机制
+            _logger.LogError(ex, "消息处理发生未知异常，连接ID：{ConnectionId}，错误：{Error}", 
+                connectionId, ex.Message);
+            
+            var success = await _errorHandler.HandleWithRetryAsync(async () =>
+            {
+                // 这里可以添加重试逻辑，但对于缓冲区问题，重试通常无效
+                throw ex; // 重新抛出异常
+            });
+
+            if (!success)
+            {
+                await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
+            }
         }
     }
 

src/modify.md

@@ -112,4 +112,330 @@
 - 提高并发处理能力，避免连接间的数据竞争
 - 减少内存泄漏风险，提高系统稳定性
 - 优化超时检测机制，及时释放无效连接
-- 改进异常处理，减少系统崩溃概率 
+- 改进异常处理，减少系统崩溃概率
+
+## 2024-12-19 WebSocket 缓冲区管理问题修复
+
+### 修改文件
+- `X1.WebSocket/Middleware/WebSocketMiddleware.cs`
+
+### 问题描述
+从日志中发现缓冲区溢出错误：
+```
+[17:19:07 WRN] 消息缓冲区已满，连接ID：ff70aebb-c0a2-40a3-b478-8aaa00daf50f，缓冲区大小：1048576，消息大小：4096
+[17:19:07 WRN] 操作失败，重试 1/3
+System.Net.WebSockets.WebSocketException (0x80004005): 消息缓冲区溢出
+```
+
+### 根本原因分析
+
+#### 1. 缓冲区大小设置错误
+- **问题**：缓冲区大小设置为 `_options.MaxMessageSize`（1MB），但这是用来累积分片消息的
+- **影响**：当消息分片传输时，缓冲区会累积所有分片直到达到最大大小，导致溢出
+
+#### 2. 分片消息处理逻辑缺失
+- **问题**：没有正确处理大型消息的分片传输
+- **影响**：分片消息无法正确累积和重组
+
+#### 3. 缓冲区管理逻辑错误
+- **问题**：在写入数据前没有检查总大小是否会超过限制
+- **影响**：可能导致缓冲区溢出
+
+### 修复方案
+
+#### 1. 优化缓冲区大小设置
+- **修复**：将缓冲区大小设置为 `MaxMessageSize` 的 1.5 倍
+- **原因**：确保有足够空间处理分片消息，避免溢出
+- **代码**：
+```csharp
+var bufferSize = (int)(_options.MaxMessageSize * 1.5);
+using var messageBuffer = new WebSocketMessageBuffer(bufferSize);
+```
+
+#### 2. 改进缓冲区写入逻辑
+- **修复**：在写入前检查总大小是否会超过限制
+- **原因**：提前检测溢出风险，避免不必要的写入操作
+- **代码**：
+```csharp
+var totalSize = messageBuffer.Size + receiveResult.Count;
+if (totalSize > _options.MaxMessageSize)
+{
+    throw new WebSocketException("消息大小超过限制");
+}
+```
+
+#### 3. 增强分片消息处理
+- **修复**：添加分片消息的详细日志记录
+- **原因**：便于调试和监控分片消息的处理过程
+- **代码**：
+```csharp
+_logger.LogDebug("消息分片已累积，连接ID：{ConnectionId}，当前缓冲区大小：{BufferSize}字节", 
+    connectionId, messageBuffer.Size);
+```
+
+#### 4. 改进错误处理和日志
+- **修复**：提供更详细的错误信息和调试日志
+- **原因**：便于问题排查和性能监控
+- **改进**：
+  - 区分"消息大小超过限制"和"缓冲区写入失败"
+  - 添加缓冲区状态和分片处理过程的详细日志
+  - 在关键操作点添加调试日志
+
+### 技术细节
+
+#### 1. 缓冲区大小计算
+- 原始大小：`_options.MaxMessageSize`（1MB）
+- 新大小：`_options.MaxMessageSize * 1.5`（1.5MB）
+- 原因：为分片消息提供足够的缓冲空间
+
+#### 2. 大小检查策略
+- 写入前检查：`totalSize = currentSize + newDataSize`
+- 最终验证：`messageData.Length > _options.MaxMessageSize`
+- 双重保护：确保消息大小在合理范围内
+
+#### 3. 分片处理流程
+- 累积分片：`messageBuffer.TryWrite(buffer, 0, receiveResult.Count)`
+- 检查完整性：`receiveResult.EndOfMessage`
+- 重组消息：`messageBuffer.GetMessage()`
+- 重置缓冲区：`messageBuffer.Reset()`
+
+### 性能影响
+- **内存使用**：缓冲区大小增加 50%，但避免了溢出错误
+- **处理效率**：提前检测大小限制，避免无效的写入操作
+- **稳定性**：正确处理分片消息，提高系统稳定性
+- **可维护性**：详细的日志记录，便于问题排查
+
+### 测试建议
+1. 测试大型消息的分片传输
+2. 验证缓冲区大小限制的正确性
+3. 检查分片消息的完整重组
+4. 监控内存使用情况
+5. 验证错误处理的正确性
+
+## 2024-12-19 WebSocket 错误处理优化
+
+### 修改文件
+- `X1.WebSocket/Middleware/WebSocketMiddleware.cs`
+
+### 问题描述
+从日志中发现重试机制在缓冲区满的情况下无效：
+```
+[17:19:07 WRN] 消息缓冲区已满，连接ID：ff70aebb-c0a2-40a3-b478-8aaa00daf50f，缓冲区大小：1048576，消息大小：4096
+[17:19:07 WRN] 操作失败，重试 1/3
+[17:19:08 WRN] 操作失败，重试 2/3
+[17:19:09 WRN] 操作失败，重试 3/3
+[17:19:09 ERR] 消息处理失败，连接ID：ff70aebb-c0a2-40a3-b478-8aaa00daf50f，关闭连接
+```
+
+### 根本原因分析
+
+#### 1. 重试机制设计缺陷
+- **问题**：对于缓冲区满等状态性问题，重试机制没有意义
+- **影响**：浪费系统资源，延迟错误处理
+
+#### 2. 异常类型区分不足
+- **问题**：没有区分不同类型的异常，所有异常都进行重试
+- **影响**：对不可恢复的错误进行无效重试
+
+#### 3. 资源清理不完整
+- **问题**：存在未使用的成员变量 `_messageBuffer`
+- **影响**：代码冗余，可能造成混淆
+
+### 修复方案
+
+#### 1. 优化异常处理策略
+- **修复**：区分 WebSocket 异常和其他异常
+- **原因**：WebSocket 异常通常不可恢复，不需要重试
+- **代码**：
+```csharp
+catch (WebSocketException ex)
+{
+    // 对于 WebSocket 异常，直接抛出，不进行重试
+    _logger.LogError(ex, "WebSocket 消息处理异常，连接ID：{ConnectionId}，错误：{Error}", 
+        connectionId, ex.Message);
+    throw;
+}
+catch (Exception ex)
+{
+    // 对于其他异常，使用重试机制
+    // ...
+}
+```
+
+#### 2. 移除未使用的成员变量
+- **修复**：移除 `_messageBuffer` 成员变量和初始化代码
+- **原因**：清理代码冗余，避免混淆
+- **改进**：
+  - 移除成员变量声明：`private readonly WebSocketMessageBuffer _messageBuffer;`
+  - 移除初始化代码：`_messageBuffer = new WebSocketMessageBuffer(_options.MaxMessageSize);`
+
+#### 3. 改进错误日志记录
+- **修复**：提供更详细的错误分类和日志信息
+- **原因**：便于问题排查和性能监控
+- **改进**：
+  - 区分 WebSocket 异常和其他异常
+  - 提供更具体的错误描述
+  - 减少无效的重试日志
+
+### 技术细节
+
+#### 1. 异常分类策略
+- **WebSocket 异常**：直接抛出，不重试
+  - 缓冲区溢出
+  - 消息大小超限
+  - 连接状态异常
+- **其他异常**：使用重试机制
+  - 网络临时错误
+  - 系统资源临时不足
+
+#### 2. 重试逻辑优化
+- **适用场景**：临时性、可恢复的错误
+- **不适用场景**：状态性、不可恢复的错误
+- **重试策略**：指数退避，避免系统过载
+
+#### 3. 资源管理改进
+- **成员变量清理**：移除未使用的 `_messageBuffer`
+- **内存优化**：减少不必要的对象创建
+- **代码简化**：提高代码可读性和维护性
+
+### 性能影响
+- **响应速度**：减少无效重试，提高错误响应速度
+- **资源使用**：减少系统资源浪费
+- **稳定性**：更准确的错误处理，提高系统稳定性
+- **可维护性**：代码更清晰，便于维护和调试
+
+### 测试建议
+1. 测试缓冲区满时的错误处理
+2. 验证不同类型异常的处理策略
+3. 检查重试机制的正确性
+4. 监控系统资源使用情况
+5. 验证错误日志的准确性 
+
+## 2024-12-19 HandleWebSocketConnection 缓冲区大小修复
+
+### 修改文件
+- `X1.WebSocket/Middleware/WebSocketMiddleware.cs`
+
+### 问题描述
+- 原实现中 buffer 大小为 4KB（`1024 * 4`），与 WebSocketOptions.MaxMessageSize（如 1MB）不匹配。
+- 当接收大消息时，分片次数多，增加缓冲区溢出和性能问题风险。
+
+### 修复方案
+- 将 buffer 大小调整为 `_options.MaxMessageSize`，与最大消息大小保持一致。
+- 这样可减少分片次数，提高性能，降低溢出风险。
+
+### 代码片段
+```csharp
+// 修复前
+var buffer = ArrayPool<byte>.Shared.Rent(1024 * 4);
+// 修复后
+var buffer = ArrayPool<byte>.Shared.Rent(_options.MaxMessageSize);
+```
+
+### 影响
+- 提高了大消息处理的效率和健壮性。
+- 降低了分片累积导致的缓冲区溢出风险。
+- 代码更符合最佳实践。 
+
+## 2024-12-19 ProcessMessage 检查逻辑优化
+
+### 修改文件
+- `X1.WebSocket/Middleware/WebSocketMiddleware.cs`
+
+### 问题描述
+- `ProcessMessage` 方法中存在冗余的大小检查逻辑
+- 在已经提前检查 `totalSize` 的情况下，后续的检查理论上不应该失败
+- 日志级别使用不当，某些错误情况应该使用 Error 级别
+
+### 优化方案
+
+#### 1. 优化检查逻辑
+- 保留提前的 `totalSize` 检查作为主要验证
+- 将后续检查作为安全验证，并调整日志级别为 Error
+- 添加注释说明检查的层次和目的
+
+#### 2. 改进日志记录
+- 将不应该发生的错误情况日志级别调整为 Error
+- 添加更明确的错误描述，便于问题排查
+
+### 代码改进
+```csharp
+// 主要检查：提前验证大小
+var totalSize = messageBuffer.Size + receiveResult.Count;
+if (totalSize > _options.MaxMessageSize)
+{
+    // 正常的大小超限情况，使用 Warning 级别
+    _logger.LogWarning("消息大小将超过限制...");
+}
+
+// 安全验证：理论上不应该失败
+if (!messageBuffer.TryWrite(buffer, 0, receiveResult.Count))
+{
+    // 不应该发生的情况，使用 Error 级别
+    _logger.LogError("消息缓冲区写入失败，这不应该发生");
+}
+
+// 最终验证：双重保护
+if (messageData.Length > _options.MaxMessageSize)
+{
+    // 不应该发生的情况，使用 Error 级别
+    _logger.LogError("最终消息大小超过限制，这不应该发生");
+}
+```
+
+### 影响
+- 提高了代码的可读性和维护性
+- 更准确的日志级别分类
+- 保持了安全检查的完整性
+- 便于问题排查和调试 
+
+## 2024-12-19 移除无意义的重试逻辑
+
+### 修改文件
+- `X1.WebSocket/Middleware/WebSocketMiddleware.cs`
+
+### 问题描述
+- `ProcessMessage` 方法中的重试逻辑没有实际意义
+- 对于 WebSocketException，已经单独 catch 并直接抛出
+- 对于其他异常，重试逻辑只是简单地 `throw ex`，没有实际重试操作
+- `HandleMessageProcessingFailure` 只会在所有重试都失败后调用，但由于每次都直接抛出异常，实际上不会有任何重试
+
+### 优化方案
+- 移除无意义的重试逻辑
+- 简化异常处理流程
+- 直接调用 `HandleMessageProcessingFailure`
+
+### 代码改进
+```csharp
+// 优化前
+catch (Exception ex)
+{
+    _logger.LogError(ex, "消息处理发生未知异常，连接ID：{ConnectionId}，错误：{Error}", 
+        connectionId, ex.Message);
+    
+    var success = await _errorHandler.HandleWithRetryAsync(async () =>
+    {
+        // 这里可以添加重试逻辑，但对于缓冲区问题，重试通常无效
+        throw ex; // 重新抛出异常
+    });
+
+    if (!success)
+    {
+        await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
+    }
+}
+
+// 优化后
+catch (Exception ex)
+{
+    _logger.LogError(ex, "WebSocket 消息处理发生未知异常，连接ID：{ConnectionId}，错误：{Error}", 
+        connectionId, ex.Message);
+    await HandleMessageProcessingFailure(webSocket, connectionId, cancellationToken);
+}
+```
+
+### 影响
+- 简化了代码逻辑，提高了可读性
+- 减少了不必要的重试开销
+- 保持了错误处理的完整性
+- 提高了代码执行效率