實現異步寫入時的讀取

Question

我有一個流可以同時讀取和寫入。 MSDN文檔說，只有到達流的末尾時，我才應該從Read(buffer, offset, count)方法中返回零。

由於讀寫操作是異步的，所以如果內部緩衝區爲空，讀取需要等待另一個寫入。

我正在努力寫作方法如何表示一切都已寫好。我能想到的最好的方法是Dispose()（或Close()）方法將標誌着寫作的結束，但這感覺非常錯誤。

我流類實現爲：

public class ContinuousStream : Stream 
{ 
    private readonly IProducerConsumerCollection<byte> _buffer; 

    public ContinuousStream() => _buffer = new ConcurrentQueue<byte>(); 

    public override int Read(byte[] buffer, int offset, int count) 
    { 
     var maxByteCount = offset + count > buffer.Length ? buffer.Length : count; 
     var actualBytesRead = 0; 

     for (var i = offset; i < maxByteCount && _buffer.TryTake(out var b); i++) 
     { 
      buffer[i] = b; 
      actualBytesRead++; 
     } 

     return actualBytesRead; 
    } 

    public override void Write(byte[] buffer, int offset, int count) 
    { 
     var maxByteCount = offset + count > buffer.Length ? buffer.Length : count; 
     for (var i = offset; i < maxByteCount; i++) 
     { 
      _buffer.TryAdd(buffer[i]); 
     } 
    } 
} 

爲了澄清

同時讀取和寫入發生。閱讀不應該等到所有內容在運行之前進行緩衝。我怎樣才能向這個信號流發出將要發生的所有寫作都已經發生？

Answer 1

其實，我不太明白爲什麼做你需要爲這個流，不能使用IProducerConsumerCollection單獨。許多這些參數都沒有意義。只需使用TryTake & TryAdd而不是讀取&寫在第一位。

但是你可以使用這個的AutoResetEvent：

public class ContinuousStream : Stream 
{ 
    private readonly AutoResetEvent _are = new AutoResetEvent(false); 
    private readonly IProducerConsumerCollection<byte> _buffer = new ConcurrentQueue<byte>(); 

    public override int Read(byte[] buffer, int offset, int count) 
    { 
     _are.WaitOne(); 
     var maxByteCount = offset + count > buffer.Length ? buffer.Length : count; 
     var actualBytesRead = 0; 

     for (var i = offset; i < maxByteCount && _buffer.TryTake(out var b); i++) 
     { 
      buffer[i] = b; 
      actualBytesRead++; 
     } 

     return actualBytesRead; 
    } 

    public override void Write(byte[] buffer, int offset, int count) 
    { 
     var maxByteCount = offset + count > buffer.Length ? buffer.Length : count; 
     for (var i = offset; i < maxByteCount; i++) 
     { 
      _buffer.TryAdd(buffer[i]); 
      _are.Set(); 
     } 
    } 
}