线程安全缓冲区包装流

Question

我在TcpClient 之上使用SslStream。不幸的是，`SslStream` 不支持同时从多个线程写入或读取。这就是为什么我围绕它编写了自己的包装器：

private ConcurrentQueue<byte> sendQueue;

private volatile bool oSending;

private readonly object writeLock;


public async void Write(byte[] buffer, int offset, int count)
{
  if (oSending)
  {
    lock (writeLock)
    {
      foreach (var b in buffer)
      {
        sendQueue.Enqueue(b);
      }
    }
  }
  else
  {
    oSending = true;
    await stream.WriteAsync(buffer, offset, count);
    oSending = false;

    lock (writeLock)
    {
      if (sendQueue.Count > 0)
      {
        Write(sendQueue.ToArray(), 0, sendQueue.Count);
        sendQueue = new ConcurrentQueue<byte>();
      }
    }
  }
}

背后的意图如下：

1. 如果流是空闲的，立即写入流。
2. 如果流忙，则写入缓冲区。
3. 如果流从发送返回，检查队列中是否有数据，递归发送。

到目前为止，我已经尝试了几种解决方案，但似乎每次都发送了太多数据。

P.S.：我知道按字节填充队列并不好，但这只是变得又快又脏。

更新：我已根据 Dirk 的评论添加了队列的删除。

Answer 1

更新

使用TPL Dataflow：

using System.Threading.Tasks.Dataflow;

public class DataflowStreamWriter
{
    private readonly MemoryStream _stream = new MemoryStream();
    private readonly ActionBlock<byte[]> _block;

    public DataflowStreamWriter()
    {
        _block = new ActionBlock<byte[]>(
                        bytes => _stream.Write(bytes, 0, bytes.Length));
    }

    public void Write(byte[] data)
    {
        _block.Post(data);
    }
}

---

这是一种更好的生产者-消费者方法。

每当有人向您的ConcurrentStreamWriter 实例写入数据时，该数据就会被添加到缓冲区中。此方法是线程安全的，多个线程可能同时写入数据。这些是你的生产者。

然后，您有一个 consumer - 消耗缓冲区中的数据并将其写入流中。

BlockingCollection<T> 用于在生产者和消费者之间进行通信。这样，如果没有人在生产，消费者就会处于空闲状态。每当生产者启动并向缓冲区写入内容时，消费者就会醒来。

消费者被延迟初始化 - 当且仅当某些数据首次可用时才会创建。

public class ConcurrentStreamWriter : IDisposable
{
    private readonly MemoryStream _stream = new MemoryStream();
    private readonly BlockingCollection<byte> _buffer = new BlockingCollection<byte>(new ConcurrentQueue<byte>());

    private readonly object _writeBufferLock = new object();
    private Task _flusher;
    private volatile bool _disposed;

    private void FlushBuffer()
    {
        //keep writing to the stream, and block when the buffer is empty
        while (!_disposed)
            _stream.WriteByte(_buffer.Take());

        //when this instance has been disposed, flush any residue left in the ConcurrentStreamWriter and exit
        byte b;
        while (_buffer.TryTake(out b))
            _stream.WriteByte(b);
    }

    public void Write(byte[] data)
    {
        if (_disposed)
            throw new ObjectDisposedException("ConcurrentStreamWriter");

        lock (_writeBufferLock)
            foreach (var b in data)
                _buffer.Add(b);

        InitFlusher();
    }

    public void InitFlusher()
    {
        //safely create a new flusher task if one hasn't been created yet
        if (_flusher == null)
        {
            Task newFlusher = new Task(FlushBuffer);
            if (Interlocked.CompareExchange(ref _flusher, newFlusher, null) == null)
                newFlusher.Start();
        }
    }

    public void Dispose()
    {
        _disposed = true;
        if (_flusher != null)
            _flusher.Wait();

        _buffer.Dispose();
    }
}

Answer 2

1. 您正在锁定对 ConcurrentQueue<T> 的访问 - 您不需要，队列已经是线程安全的
2. if(oSending) {} else {oSending = true} 不是线程安全的。两个线程可能将oSending 读为false，进入else 块并将其设置为true。现在您有两个线程写入流。
3. 正如 Dirk 指出的那样，您并没有从队列中删除项目。

---

我的修改：

1. 不使用布尔标志，而是使用Monitor.TryEnter 来尝试访问流。如果当前正在写入流，则调用将立即返回 - 并继续写入缓冲区。

2. 实现IDisposable 并确保Dispose 刷新缓冲区。

3. 仅在写入队列时锁定队列，以保持字节顺序
4. 将签名从async void更改为async Task。

private readonly ConcurrentQueue<byte> _bufferQueue = new ConcurrentQueue<byte>();

private readonly object _bufferLock = new object();
private readonly object _streamLock = new object();
private readonly MemoryStream stream = new MemoryStream();

public async Task Write(byte[] data, int offset, int count)
{
    bool streamLockTaken = false;

    try
    {
        //attempt to acquire the lock - if lock is currently taken, return immediately
        Monitor.TryEnter(_streamLock, ref streamLockTaken);

        if (streamLockTaken) //write to stream
        {
            //write data to stream and flush the buffer
            await stream.WriteAsync(data, offset, count);
            await FlushBuffer();

        }
        else //write to buffer
        {
            lock (_bufferLock)
                foreach (var b in data)
                    _bufferQueue.Enqueue(b);
        }
    }
    finally
    {
        if (streamLockTaken)
            Monitor.Exit(_streamLock);
    }
}

private async Task FlushBuffer()
{
    List<byte> bufferedData = new List<byte>();
    byte b;
    while (_bufferQueue.TryDequeue(out b))
        bufferedData.Add(b);

    await stream.WriteAsync(bufferedData.ToArray(), 0, bufferedData.Count);
}

public void Dispose()
{
    lock(_streamLock)
        FlushBuffer().Wait();
}

Answer 3

你不能只锁定底层流吗？我相信它可以像这样简单：

private readonly object writeLock = new Object();

public async void Write(byte[] buffer, int offset, int count)
{
   lock (writeLock)
   {
      await stream.WriteAsync(buffer, offset, count);
   }
}

另外，通过您的队列实现，我认为写入可以排队的机器人永远不会写入流的变化。例如，在另一个线程出队之后但在该线程释放其锁之前排队。