C# .NET 构建高性能WebSocket服务端：从Fleck入门到实战优化

2026/6/19 22:08:41

1. WebSocket基础与Fleck入门

WebSocket协议是现代Web应用中实现实时双向通信的基石。相比传统的HTTP轮询，它就像给浏览器和服务器之间架设了一条双向高速公路——建立连接后，双方可以随时发送数据，不再需要反复"打电话确认"。我在处理在线协作白板项目时，实测WebSocket的延迟能控制在50ms以内，而HTTP轮询平均需要200ms以上。

Fleck是一个轻量级的C# WebSocket库，它的优势在于：

纯C#实现，不依赖System.Net.WebSockets
支持异步IO处理
内置心跳检测机制
代码简洁直观

安装只需在NuGet执行：

Install-Package Fleck

基础服务搭建示例：

// 连接池使用线程安全的ConcurrentBag var sockets = new ConcurrentBag<IWebSocketConnection>(); var server = new WebSocketServer("ws://0.0.0.0:8181"); server.Start(socket => { socket.OnOpen = () => { sockets.Add(socket); Console.WriteLine($"连接建立 {socket.ConnectionInfo.Id}"); }; socket.OnClose = () => { sockets.TryTake(out _); Console.WriteLine($"连接关闭 {socket.ConnectionInfo.Id}"); }; socket.OnMessage = message => { Console.WriteLine($"收到消息 {message}"); // 广播给所有客户端 Parallel.ForEach(sockets, s => { if(s.IsAvailable) s.Send($"Echo: {message}"); }); }; });

2. 连接管理与性能优化

2.1 连接池的工业级实现

原始示例中的List在高并发下会出现线程安全问题。我在实际项目中遇到过这样的场景：当500+用户同时连接时，简单的List操作会导致连接丢失。改进方案：

// 使用线程安全集合 private readonly ConcurrentDictionary<Guid, IWebSocketConnection> _activeConnections = new ConcurrentDictionary<Guid, IWebSocketConnection>(); // 添加连接时 socket.OnOpen = () => { _activeConnections.TryAdd(socket.ConnectionInfo.Id, socket); // 记录连接元数据 socket.ConnectionInfo.Headers.TryGetValue("User-Agent", out var ua); _logger.LogInformation($"新连接 {socket.ConnectionInfo.Id} from {ua}"); };

2.2 心跳检测与超时处理

Fleck内置了心跳机制，但需要手动配置：

var server = new WebSocketServer("ws://0.0.0.0:8181") { // 10秒无活动自动断开 RestartAfterListenFailure = true, ListenerSocket = { NoDelay = true } }; // 自定义心跳检测 Task.Run(async () => { while(true) { await Task.Delay(30000); var deadConnections = _activeConnections .Where(c => !c.Value.IsAvailable) .ToList(); foreach(var conn in deadConnections) { _activeConnections.TryRemove(conn.Key, out _); conn.Value.Close(); } } });

3. 消息处理与广播策略

3.1 高效消息广播模式

直接遍历连接池发送消息会遇到性能瓶颈。通过测试发现，当连接数超过1000时，简单广播的延迟会显著上升。优化方案：

// 使用消息队列+批量发送 private readonly BlockingCollection<string> _messageQueue = new BlockingCollection<string>(); // 消息生产者 socket.OnMessage = message => { _messageQueue.Add(message); }; // 消息消费者 Task.Run(() => { var buffer = new List<string>(100); foreach(var msg in _messageQueue.GetConsumingEnumerable()) { buffer.Add(msg); if(buffer.Count >= 100) { var json = JsonSerializer.Serialize(buffer); Parallel.ForEach(_activeConnections.Values, socket => { try { socket.Send(json); } catch(WebSocketException ex) { _logger.LogError(ex, "发送失败"); } }); buffer.Clear(); } } });

3.2 消息压缩与二进制传输

文本消息超过1KB时建议启用压缩：

using(var ms = new MemoryStream()) { using(var gzip = new GZipStream(ms, CompressionMode.Compress)) { var bytes = Encoding.UTF8.GetBytes(message); gzip.Write(bytes, 0, bytes.Length); } socket.Send(ms.ToArray()); }

4. 监控与异常处理

4.1 关键指标监控

在.NET中集成Prometheus监控：

private static readonly Counter _messagesReceived = Metrics .CreateCounter("websocket_messages_total", "接收消息总数"); private static readonly Gauge _activeConnectionsGauge = Metrics .CreateGauge("websocket_connections_active", "活跃连接数"); socket.OnMessage = message => { _messagesReceived.Inc(); // ...处理逻辑 }; // 定时更新连接数指标 Task.Run(async () => { while(true) { _activeConnectionsGauge.Set(_activeConnections.Count); await Task.Delay(5000); } });

4.2 异常处理最佳实践

WebSocket常见异常包括：

连接意外中断
消息过大导致内存溢出
网络抖动造成的超时

全局异常处理示例：

socket.OnError = ex => { _logger.LogError(ex, $"连接异常 {socket.ConnectionInfo.Id}"); if(ex is WebSocketException wsEx) { socket.Close((int)wsEx.StatusCode); } else { socket.Close(500); } }; // 配置全局未处理异常捕获 AppDomain.CurrentDomain.UnhandledException += (sender, args) => { _logger.LogCritical(args.ExceptionObject as Exception, "全局异常"); };

5. 负载均衡与水平扩展

当单机性能达到瓶颈时（通常约5000-10000连接），需要考虑水平扩展。我在实际项目中采用Redis作为共享状态存储：

// 使用Redis发布订阅 var redis = ConnectionMultiplexer.Connect("localhost"); var sub = redis.GetSubscriber(); // 订阅跨节点消息 sub.Subscribe("broadcast", (channel, message) => { var msg = Encoding.UTF8.GetString(message); foreach(var socket in _activeConnections.Values) { socket.Send(msg); } }); // 发送跨节点消息 socket.OnMessage = message => { sub.Publish("broadcast", message); };

6. 安全防护实践

6.1 连接认证

建议在握手阶段完成认证：

server.Start(socket => { if(!Authenticate(socket.ConnectionInfo)) { socket.Close(1008, "认证失败"); return; } // ...其他逻辑 }); private bool Authenticate(IWebSocketConnectionInfo info) { return info.Headers.TryGetValue("Authorization", out var token) && ValidateToken(token); }

6.2 消息速率限制

防止恶意用户发送大量消息：

private readonly ConcurrentDictionary<Guid, RateLimiter> _rateLimiters = new ConcurrentDictionary<Guid, RateLimiter>(); socket.OnMessage = message => { var limiter = _rateLimiters.GetOrAdd( socket.ConnectionInfo.Id, _ => new RateLimiter(10, TimeSpan.FromSeconds(1)) ); if(!limiter.TryAcquire()) { socket.Close(1008, "发送频率过高"); return; } // ...处理消息 };

7. 性能调优实战

经过多次压力测试，总结出这些关键配置：

// 调整线程池设置 ThreadPool.SetMinThreads(100, 100); ThreadPool.SetMaxThreads(32767, 32767); // 优化GC行为 GCSettings.LatencyMode = GCLatencyMode.SustainedLowLatency; // 服务端配置 var server = new WebSocketServer("ws://0.0.0.0:8181") { // 启用SO_REUSEADDR RestartAfterListenFailure = true, // 禁用Nagle算法 ListenerSocket = { NoDelay = true, SendTimeout = 1000, ReceiveTimeout = 1000 }, // 自定义证书验证 CertificateValidationCallback = (cert, chain, errors) => true };

在8核16G的服务器上，经过优化后可以稳定支持：