.NET WebSocket高并发通讯阻塞问题
项目上碰到使用WebSocket超时问题，具体情况是这样的，OTA升级过程中，解压zip文件会有解压进度事件，将解压进度通过历程通讯传给另一历程，通讯提示超时非常小伙伴堂园发现大文件使用Zip解压，解压进度事件间隔竟然是1ms，简直超大频率啊
但是，解压事件超频也不应该通讯非常啊，于是我通过1ms定时发送通讯事件，测试了下历程间通讯流程。
WebSocketSharp
当前历程间通讯组件是基于kaistseo/UnitySocketIO-WebSocketSharp实现，主机内设置一服务端，多个客户端毗连服务端，客户端通讯由服务端转发数据。客户端A发送给B后，客户端B会将实行结果反馈给客户A。
那在定位中发现，各个链路发送延时都是正常的，包括服务端发送反馈数据给到客户端A，但客户端A接收数据延时很大，下面是部分返回数据：
https://img2024.cnblogs.com/blog/685541/202409/685541-20240903203724427-264007743.png
并且通讯时间久了之后，延时会越来越大
这里是WebSocketSharp.WebSocket对外事件OnMessage：
1 private void WebSocketOnMessage(object sender, MessageEventArgs e)
2 {
3 if (!e.IsText)
4 {
5 //暂时不支持
6 return;
7 }
8 Debug.WriteLine($"{DateTime.Now.ToString("HH:mm:ss fff")},{e.Data}");
9
10 var receivedMessage = JsonConvertSlim.Decode<ChannelServerMessage>(e.Data);
11 xxxxx
12 }我们继承往下看，OnMessage是由WebSocket.message()触发，从_messageEventQueue队列中获取数据：
1 private void message ()
2 {
3 MessageEventArgs e = null;
4 lock (_forMessageEventQueue) {
5 if (_inMessage || _messageEventQueue.Count == 0 || _readyState != WebSocketState.Open)
6 return;
7
8 _inMessage = true;
9 e = _messageEventQueue.Dequeue ();
10 }
11
12 _message (e);
13 }循环接收数据是这样拿的：
1 private void startReceiving ()
2 {
3 xxxx
4 _receivingExited = new ManualResetEvent (false);
5 Action receive = () => WebSocketFrame.ReadFrameAsync (_stream, false,
6 frame => {
7 if (!processReceivedFrame (frame) || _readyState == WebSocketState.Closed) {
8 var exited = _receivingExited;
9 if (exited != null)
10 exited.Set ();
11 return;
12 }
13 // Receive next asap because the Ping or Close needs a response to it.
14 receive ();
15 xxxx
16 message ();
17 },
18 xxxx
19 );
20 receive ();
21 }这里我看到了ManualResetEvent。。。数据量那么大，这里搞个同步信号锁，肯定会堵住咯
为何设置线程同步锁呢？我们往下看
WebSocketSharp数据发送是基于TCPClient实现的：
1 _tcpClient = new TcpClient (_proxyUri.DnsSafeHost, _proxyUri.Port);
2 _stream = _tcpClient.GetStream ();初始化后通过_stream.Write (bytes, 0, bytes.Length);发送数据
接收数据，也是通过_stream读取，可以看上方的startReceiving()方法里，WebSocketFrame.ReadFrameAsync (_stream, false,...)
我们知道，TCP是面向毗连，提供可靠、顺序的数据流传输。用于一对一的通讯，即一个TCP毗连只能有一个发送方和一个接收方。具体的可以看我之前写的文章：.NET TCP、UDP、Socket、WebSocket - 唐宋元明清2188 - 博客园 (cnblogs.com)
但接收时在高并发场景下，得当的同步措施依然是必须的。我们可以使用lock也可以用SemaphoreSlim来实现复杂的同步需求，这里使用的是信号锁ManualResetEvent
我们再看看发送端代码，也是用了lock一个object来限制并发操作：
1 private bool send (Opcode opcode, Stream stream)
2 {
3 lock (_forSend) {
4 var src = stream;
5 var compressed = false;
6 var sent = false;
7 xxxxx
8 sent = send (opcode, stream, compressed);
9 xxxxx
10 return sent;
11 }
12 }以是WebSocketSharp在高并发场景下是存在通讯阻塞问题的。当然，WebSocketSharp已经实现的很好了，正常的话几ms都不会碰到阻塞问题，如下设置3ms定时超频发送、发送一段时间后：
https://img2024.cnblogs.com/blog/685541/202409/685541-20240904180640711-875029456.png
客户端A发送消息，由服务端转发至客户B，再将客户端B的反馈结果由服务端转发回客户端A，真正延时才0-2ms！
以是上方项目中碰到的ZIP文件解压进度超快1ms，只能要ZIP解压处优化下，设置并发操作10ms内保存末了一个操作，可以参考 .NET异步并发操作，只保存末了一次操作 - 唐宋元明清2188 - 博客园 (cnblogs.com)，即10ms最多触发一次解压进度事件。确实也应该这么优化，通讯即使撑住这种高并发，UI革新这么高帧率也有点浪费CPU/GPU资源。
WebSocket
我们再看看原生的WebSocket，写个WebSocket通讯Demo kybs00/WebSocketDemo (github.com)
服务端定时1ms使劲往客户端发送Message消息，结果竟然是：
System.InvalidOperationException:“There is already one outstanding 'SendAsync' call for this WebSocket instance. ReceiveAsync and SendAsync can be called simultaneously, but at most one outstanding operation for each of them is allowed at the same time.”
看来发送事件外部也要处置惩罚好高并发的场景，1ms真的是太猛了
1 private SemaphoreSlim _sendLock = new SemaphoreSlim(1);
2 private async void Timer_Elapsed(object sender, ElapsedEventArgs e)
3 {
4 var message = $"{DateTime.Now.ToString("HH:mm:ss fff")},hello from server";
5
6 await _sendLock.WaitAsync();
7 await BroadcastAsync("test", message);
8 _sendLock.Release();
9 Console.WriteLine(message);
10 }加完信号量同步，服务端就能正常发送了。下面是10分钟后客户端接收数据打印，传输几乎无延时：
https://img2024.cnblogs.com/blog/685541/202409/685541-20240904172559638-492767071.png
另外，也尝试了单独在客户端接收添加信号量同步，依然是提示服务端发送不支持并行的非常。
以是原生WebSocket在发送端需要串行处置惩罚，写入完数据后再实行_stream.FlushAsync()。
出处：http://www.cnblogs.com/kybs0/本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须在文章页面给出原文毗连，否则保存追究法律责任的权利。
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]