.NET Framework多任务调用ClientWebSocket SendAsync/ReceiveAsync异常

基于System.Net.WebSockets库实现自定义WebSocket客户端，内部通过两个独立Task分别处理命令发送和响应接收，均采用后台无限循环逻辑。因使用.NET Framework 4.5.2无官方Channel组件，自行实现了类似Channel的队列类用于commandQueue和responseQueue，目的是高效无阻塞地在SendTask中写入responseQueue、在ReceiveTask中读取。

核心代码实现 #

SendTask代码 #

// Task that sends command through the web socket and write them to response queue.
// Implementation of responseQueue & commandQueue are below. 
// CommandQueue has some items at the beginning.
while (webSocket.State != WebSocketState.Closed)
{
  var command = await commandQueue.ReadAsync(cts);
  await webSocket.SendAsync(bytesToSend, WebSocketMessageType.Text, true, CancellationToken.None);
  //in order to log command sent, push it into response queue.
  responseQueue.Push(command);
}

ReceiveTask代码 #

//Task that receive responses from the same web socket and push them into response queue as well.

//these variables needs for a multipacket responses
int dataPerPacket = 8 * 1024; //8192
StringBuilder strBuilder = new StringBuilder();
var buffer = new ArraySegment<byte>(new byte[dataPerPacket]);

while (webSocket.State != WebSocketState.Closed)
{
  var receiveResult = await webSocket.ReceiveAsync(buffer, recCts);

  // receive message
  strBuilder.Append(Encoding.UTF8帮助 Des，//搭建 detailed来自.ch来自(比对 
  strBuilder.Append(Encoding.UTF8.GetString(buffer.Array, buffer.Offset, receiveResult.Count));
  //keep doing until end of message
  while (!receiveResult.EndOfMessage)
  {
      receiveResult = await webSocket.ReceiveAsync(buffer, recCts);
      strBuilder.Append(Encoding.UTF8.GetString(buffer.Array, buffer.Offset, receiveResult.Count));
  }

  responseQueue.Push(strBuilder.ToString());
  strBuilder.Clear();
}

自定义Channel队列实现 #

public class Channel
{
        private ConcurrentQueue<string> _queue;
        private SemaphoreSlim _flag;

        public Channel()
        {
            _queue = new ConcurrentQueue<string>();
            _flag = new SemaphoreSlim(0);
        }
        
        public async Task<string> ReadAsync(CancellationToken readCancelToken)
        {
            await _flag.WaitAsync(readCancelToken);
            if (readCancelToken.IsCancellationRequested)
            {
                return default;
            }

            if (_queue.TryDequeue(out string item))
            {
                return item;
            }

            return default;
        }

        public void Push(string item)
        {
            _queue.Enqueue(item);
            _flag.Release();
        }
    }

异常现象 #

预期responseQueue的推入顺序为命令与响应交替（如command1、command2、response1、command3、response2...），实际却是先推入所有命令（如command1到command100），再批量推入所有响应。

已完成的排查 #

• 在SendTask中添加await Task.Delay(1)后，可得到预期的交替顺序
• 用Stopwatch测量SendAsync耗时为0毫秒
• 查看WebSocketClient源码确认SendAsync应存在耗时

原因分析 #

这是异步方法同步完成导致的线程调度问题：
在.NET Framework 4.5.2中，WebSocket.SendAsync在数据量小、底层传输缓冲区未填满时，会直接同步完成操作，不会触发线程切换。此时SendTask会持续占用当前线程，快速遍历commandQueue中的所有命令，完成发送并推入responseQueue。而ReceiveTask由于没有线程调度机会，只能等到SendTask循环暂时空闲（比如commandQueue为空）时，才能开始处理服务器响应，最终导致所有命令先入队、再批量入队响应的现象。

添加await Task.Delay(1)会强制触发线程上下文切换，让ReceiveTask获得调度执行的机会，从而实现命令与响应的交替入队。

解决方法 #

方法1：强制让出线程（简单有效） #

在SendTask中每次推入命令到responseQueue后，添加await Task.Yield()，主动让出当前线程，让ReceiveTask有机会被调度：

await web.Send火车 Specifically- sim要求await 直�/DOCX 
await webSocket.SendAsync(bytesToSend, WebSocketMessageType.Text, true, CancellationToken.None);
responseQueue.Push(command);
await Task.Yield(); // 让出线程，给ReceiveTask执行机会

方法2：确保SendAsync异步执行 #

将发送操作包装到Task.Run中，强制在后台线程执行，避免SendTask持续占用当前线程：

await Task.Run(() => webSocket.SendAsync(bytesToSend, WebSocketMessageType.Text, true, CancellationToken.None));
responseQueue.Push(command);

注意：需确认ClientWebSocket实例的线程安全性，System.Net.WebSockets.ClientWebSocket在.NET Framework中是线程安全的，但仍需谨慎操作。

方法3：业务层面控制发送节奏 #

如果业务允许，可以改为等待前一个命令的响应返回后再发送下一个命令，这种方式会改变原有的并发发送逻辑，但能严格保证命令与响应的顺序对应：

// 修改SendTask逻辑，每次发送后等待对应响应
while (webSocket.State != WebSocketState.Closed)
{
  var command = await commandQueue.ReadAsync(cts);
  await webSocket.SendAsync(bytesToSend, WebSocketMessageType.Text, true, CancellationToken.None);
  responseQueue.Push(command);
  
  // 等待服务器返回对应响应（需根据业务标识匹配）
  var response = await responseQueue.ReadAsync(cts);
  // 处理响应...
}

内容的提问来源于stack exchange，提问作者Sergey Skrynnik