C# .NET Core WebSocket写入操作挂起错误的解决咨询

错误原因 #

你遇到的"This method may not be called when another write operation is pending"错误，本质是**SslStream不支持并发异步写操作**。当客户端数量或数据流增加时，多个NetStreamSendAsync调用会同时触发WriteAsync，此时底层流会检测到存在未完成的写操作，从而抛出该异常。

解决方案 #

由于你的场景要求新数据覆盖旧数据（无需队列存储历史数据），我们可以通过异步锁+当前待发送数据缓存的方式解决：

• 用SemaphoreSlim实现异步互斥，确保同一时间只有一个写操作执行；
• 用变量存储当前待发送的数据，新数据直接覆盖旧数据，避免无效的历史数据发送。

修改后的完整代码 #

// 在类中添加以下私有字段
private readonly SemaphoreSlim _sendLock = new SemaphoreSlim(1, 1);
private byte[] _currentPendingBuffer;

private async Task<int> NetStreamSendAsync(byte[] buffers, OpCode opCode = OpCode.Text)
{
    try
    {
        if (!this.Active) return -2;
        if (!this.Connected)
        {
            this.OnClientError?.Invoke(this, this.EndPoint, new SocketException((int)SocketError.NotConnected));
            return -1;
        }
        if ((opCode == OpCode.Text || opCode == OpCode.Binary) && (buffers == null || buffers.Length == 0)) return 0;
        
        var stream = this.GetSslStream();
        if (stream == null) return -1;

        // 准备最终要发送的数据包（WebSocket打包处理）
        byte[] preparedBuffer = buffers;
        if (this.ConnectionType == ConnectionType.WebSocket)
        {
            using (var packet = new WebSocketPacket(this))
                preparedBuffer = packet.Packet(buffers, opCode);
        }
        else if (preparedBuffer == null || preparedBuffer.Length == 0)
        {
            return 0;
        }

        // 用新数据覆盖待发送的旧数据
        _currentPendingBuffer = preparedBuffer;
        var bufferToSend = preparedBuffer;

        try
        {
            await _sendLock.WaitAsync(this.CancelToken.Token).ConfigureAwait(false);
            
            // 检查当前待发送数据是否被新数据替换，如果已替换则跳过本次发送
            if (_currentPendingBuffer != bufferToSend)
            {
                return bufferToSend.Length;
            }

            // 执行发送操作
            await stream.WriteAsync(bufferToSend, 0, bufferToSend.Length, this.CancelToken.Token).ConfigureAwait(false);
            await stream.FlushAsync(this.CancelToken.Token).ConfigureAwait(false);

            this.LastMessageTime = DateTime.Now;
            // 发送完成后清空待发送缓存
            _currentPendingBuffer = null;
        }
        catch (Exception e)
        {
            LogHelper.Debug($"{SkipData++} - Error: {e.Message} - Veri: {Encoding.GetString(bufferToSend)}");
            this.OnClientError?.Invoke(this, this.EndPoint, new SocketException((int)SocketError.SocketError));
        }
        finally
        {
            _sendLock.Release();
        }

        return bufferToSend.Length;
    }
    catch (Exception e)
    {
        LogHelper.Debug($"[NetStreamSendAsync] Error: {e}");
        Console.WriteLine($"[NetStreamSendAsync] Error: {e}");
        throw;
    }
}

关键改动说明 #

1. 异步锁SemaphoreSlim：

   • 用WaitAsync获取锁，确保同一时间只有一个线程/异步任务执行写操作，彻底避免并发写冲突；
   • 在finally块中释放锁，保证即使发生异常也不会导致锁永久占用。

2. 数据覆盖逻辑：

   • 每次准备好发送数据后，直接将_currentPendingBuffer替换为新数据；
   • 获取锁后先检查当前待发送数据是否还是自己准备的版本，如果不是，说明已经有新数据覆盖了，直接跳过本次发送，避免发送过时数据。

3. 简化冗余代码：

   • 去掉了不必要的await Task.FromResult，直接返回对应数值，提升代码简洁性。

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