是否有任何 .NET 数据结构/类组合允许将字节数据附加到缓冲区的末尾,但所有查看和读取都是从头开始的,从而在我读取时缩短了缓冲区?
MemoryStream
类似乎完成了其中的一部分,但我需要为读取和写入维护不同的位置,并且它不会在读取数据后自动丢弃数据。
已在对 this question 的回复中发布了答案这基本上是我正在尝试做的,但我更喜欢我可以在同一进程的不同组件中进行异步 I/O 的操作,就像普通管道甚至网络流一样(我需要过滤/处理数据优先)。
最佳答案
我将发布一次我为工作项目编写的一些逻辑的精简副本。这个版本的优点是它与缓冲数据的链接列表一起工作,因此您不必在读取时缓存大量内存和/或复制内存。此外,它的线程安全并且表现得像网络流,即:当没有可用数据时读取:WAITING直到有可用数据或超时。另外,当读取x个字节而只有y个字节时,读取所有字节后返回。希望对您有所帮助!
public class SlidingStream : Stream
{
#region Other stream member implementations
...
#endregion Other stream member implementations
public SlidingStream()
{
ReadTimeout = -1;
}
private readonly object _writeSyncRoot = new object();
private readonly object _readSyncRoot = new object();
private readonly LinkedList<ArraySegment<byte>> _pendingSegments = new LinkedList<ArraySegment<byte>>();
private readonly ManualResetEventSlim _dataAvailableResetEvent = new ManualResetEventSlim();
public int ReadTimeout { get; set; }
public override int Read(byte[] buffer, int offset, int count)
{
if (_dataAvailableResetEvent.Wait(ReadTimeout))
throw new TimeoutException("No data available");
lock (_readSyncRoot)
{
int currentCount = 0;
int currentOffset = 0;
while (currentCount != count)
{
ArraySegment<byte> segment = _pendingSegments.First.Value;
_pendingSegments.RemoveFirst();
int index = segment.Offset;
for (; index < segment.Count; index++)
{
if (currentOffset < offset)
{
currentOffset++;
}
else
{
buffer[currentCount] = segment.Array[index];
currentCount++;
}
}
if (currentCount == count)
{
if (index < segment.Offset + segment.Count)
{
_pendingSegments.AddFirst(new ArraySegment<byte>(segment.Array, index, segment.Offset + segment.Count - index));
}
}
if (_pendingSegments.Count == 0)
{
_dataAvailableResetEvent.Reset();
return currentCount;
}
}
return currentCount;
}
}
public override void Write(byte[] buffer, int offset, int count)
{
lock (_writeSyncRoot)
{
byte[] copy = new byte[count];
Array.Copy(buffer, offset, copy, 0, count);
_pendingSegments.AddLast(new ArraySegment<byte>(copy));
_dataAvailableResetEvent.Set();
}
}
}
关于c# - 专用于字节流的FIFO/Queue buffer,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/8221136/