我的应用程序从TCP套接字中读取字节并需要对其进行缓冲，以便稍后可以从中提取消息。由于TCP的特性，我可能会在一次读取中获得部分或多个消息，因此每次读取后，我想检查缓冲区并提取尽可能多的完整消息。
因此，我希望有一个类能够让我执行以下操作：
我期望我想要的可以使用.NET库中的1个或多个现有类来实现，但我不确定哪些类。 System.IO.MemoryStream看起来接近我想要的，但是(a)不清楚它是否适合用作缓冲区（读取的数据是否从容量中删除？）和(b)读取和写入似乎发生在同一个位置-"流的当前位置是下一次读取或写入操作可以发生的位置。"-这不是我想要的。我需要从前面写入并从前面读取。
我建议你在内部使用MemoryStream，但是将其封装在另一个类中，该类存储：
MemoryStream然后该类会公开以下内容：
MemoryStream中并更新所有变量。（你可能不希望在每个消费请求上都复制缓冲区。）请注意，如果没有额外的同步，这些操作都不是线程安全的。
MemoryStream相较于普通的byte[]数组的优点是什么？（好了，算了，我知道了：它可以自动调整大小） - vgruMemoryStream具有可扩展的容量（如果需要）。 - OliverMemoryStream.GetBuffer获取底层缓冲区，然后使用普通的Write调用将一部分写入新流中。 - Jon Skeet只需使用一个大的字节数组和 Array.Copy - 就可以搞定了。
如果不行，就使用 List<byte>。
如果你使用数组，你需要自己实现一个索引（用于复制额外数据），以及检查内容大小，但这很简单。
如果你感兴趣，这里有一个“循环缓冲区”的简单实现。测试应该会运行（我对它进行了一些单元测试，但没有检查所有关键路径）：
public class ReadWriteBuffer
{
private readonly byte[] _buffer;
private int _startIndex, _endIndex;
public ReadWriteBuffer(int capacity)
{
_buffer = new byte[capacity];
}
public int Count
{
get
{
if (_endIndex > _startIndex)
return _endIndex - _startIndex;
if (_endIndex < _startIndex)
return (_buffer.Length - _startIndex) + _endIndex;
return 0;
}
}
public void Write(byte[] data)
{
if (Count + data.Length > _buffer.Length)
throw new Exception("buffer overflow");
if (_endIndex + data.Length >= _buffer.Length)
{
var endLen = _buffer.Length - _endIndex;
var remainingLen = data.Length - endLen;
Array.Copy(data, 0, _buffer, _endIndex, endLen);
Array.Copy(data, endLen, _buffer, 0, remainingLen);
_endIndex = remainingLen;
}
else
{
Array.Copy(data, 0, _buffer, _endIndex, data.Length);
_endIndex += data.Length;
}
}
public byte[] Read(int len, bool keepData = false)
{
if (len > Count)
throw new Exception("not enough data in buffer");
var result = new byte[len];
if (_startIndex + len < _buffer.Length)
{
Array.Copy(_buffer, _startIndex, result, 0, len);
if (!keepData)
_startIndex += len;
return result;
}
else
{
var endLen = _buffer.Length - _startIndex;
var remainingLen = len - endLen;
Array.Copy(_buffer, _startIndex, result, 0, endLen);
Array.Copy(_buffer, 0, result, endLen, remainingLen);
if (!keepData)
_startIndex = remainingLen;
return result;
}
}
public byte this[int index]
{
get
{
if (index >= Count)
throw new ArgumentOutOfRangeException();
return _buffer[(_startIndex + index) % _buffer.Length];
}
}
public IEnumerable<byte> Bytes
{
get
{
for (var i = 0; i < Count; i++)
yield return _buffer[(_startIndex + i) % _buffer.Length];
}
}
}
BufferedStream是解决问题的方法。而且可以通过调用Seek跳过未读取的len字节数据。BufferdStream buffer = new BufferedStream(tcpStream, size); // we have a buffer of size
...
...
while(...)
{
buffer.Read(...);
// do my staff
// I read too much, I want to put back len bytes
buffer.Seek(-len, SeekOrigin.End);
// I shall come back and read later
}
MemoryStream可以自动增长，与BufferedStream不同的是，BufferedStream需要初始指定size。
MemoryStream一直保存所有的数据，而BufferedStream只保存一段流数据。
MemoryStream允许使用Write()方法添加输入字节，这些字节可以在未来用Read()方法读取。而BufferedSteam从构造函数中指定的另一个源流中获取输入字节。
来晚了，但为了后人记录:
过去我做过这件事，采用了稍微不同的方法。 如果 消息有一个固定的头部大小（告诉您正文有多少字节），并且考虑到网络流已经在缓冲，我将操作分为两个阶段：
这利用了流的特点——当您请求'n'个字节时，您永远不会得到多余的字节，因此您可以忽略许多“哦，我读得太多了，请让我把它们放在下次读取”等问题。
现在这还不是全部，公平地说。我在流上有一个底层包装器类来处理碎片化问题（即如果要求4个字节，直到收到4个字节或流关闭才返回）。但那很容易。
在我看来，关键是将消息处理与流机制解耦，如果停止尝试从流中作为单个ReadBytes()来消耗消息，生活就变得简单得多。
无论您的读取是阻塞还是异步（APM/await），所有这些都是真实的。这是我写了一段时间的缓冲区的另一个实现：
听起来你想从套接字中读取数据到一个内存流缓冲区，然后每次遇到特定的字节时，从缓冲区中取出数据并重置它。代码大致如下：
void ReceiveAllMessages(Action<byte[]> messageReceived, Socket socket)
{
var currentMessage = new MemoryStream();
var buffer = new byte[128];
while (true)
{
var read = socket.Receive(buffer, 0, buffer.Length);
if (read == 0)
break; // Connection closed
for (var i = 0; i < read; i++)
{
var currentByte = buffer[i];
if (currentByte == END_OF_MESSAGE)
{
var message = currentMessage.ToByteArray();
messageReceived(message);
currentMessage = new MemoryStream();
}
else
{
currentMessage.Write(currentByte);
}
}
}
}
你可以使用一个包装了ConcurrentQueue<ArraySegment<byte>>的Stream来实现这个功能（请记住，这使其仅支持向前）。然而，我真的不喜欢在处理数据之前将数据保存在内存中的想法；这会让你面临一堆攻击（有意或无意）关于消息大小。你可能还想Google“循环缓冲区”。
实际上，你应该编写代码，尽快对接收到的数据执行有意义的操作：“推送解析”（例如，SAX就支持这种方式）。以下是如何使用文本进行此操作的示例：
private Encoding _encoding;
private Decoder _decoder;
private char[] _charData = new char[4];
public PushTextReader(Encoding encoding)
{
_encoding = encoding;
_decoder = _encoding.GetDecoder();
}
// A single connection requires its own decoder
// and charData. That connection should never
// call this method from multiple threads
// simultaneously.
// If you are using the ReadAsyncLoop you
// don't need to worry about it.
public void ReceiveData(ArraySegment<byte> data)
{
// The two false parameters cause the decoder
// to accept 'partial' characters.
var charCount = _decoder.GetCharCount(data.Array, data.Offset, data.Count, false);
charCount = _decoder.GetChars(data.Array, data.Offset, data.Count, _charData, 0, false);
OnCharacterData(new ArraySegment<char>(_charData, 0, charCount));
}
如果您必须在反序列化之前能够接受完整的消息，可以使用MemoryMappedFile，它的优点是发送实体无法使您的服务器出现内存不足的情况。问题在于将文件重置为零；因为这可能会引起一系列问题。解决此问题的一种方法是：
TCP接收端
反序列化端
TCP接收端非常简单。反序列化端将需要一些基本的缓冲区拼接逻辑（记得使用Buffer.BlockCopy而不是Array.Copy）。
附注：听起来像一个有趣的项目，如果我有时间并且记得，我可能会实现这个系统。
这里只有三个回答提供了代码。 其中一个笨拙，其他的都没有回答问题。
这是一个你可以直接复制粘贴的类：
/// <summary>
/// This class is a very fast and threadsafe FIFO buffer
/// </summary>
public class FastFifo
{
private List<Byte> mi_FifoData = new List<Byte>();
/// <summary>
/// Get the count of bytes in the Fifo buffer
/// </summary>
public int Count
{
get
{
lock (mi_FifoData)
{
return mi_FifoData.Count;
}
}
}
/// <summary>
/// Clears the Fifo buffer
/// </summary>
public void Clear()
{
lock (mi_FifoData)
{
mi_FifoData.Clear();
}
}
/// <summary>
/// Append data to the end of the fifo
/// </summary>
public void Push(Byte[] u8_Data)
{
lock (mi_FifoData)
{
// Internally the .NET framework uses Array.Copy() which is extremely fast
mi_FifoData.AddRange(u8_Data);
}
}
/// <summary>
/// Get data from the beginning of the fifo.
/// returns null if s32_Count bytes are not yet available.
/// </summary>
public Byte[] Pop(int s32_Count)
{
lock (mi_FifoData)
{
if (mi_FifoData.Count < s32_Count)
return null;
// Internally the .NET framework uses Array.Copy() which is extremely fast
Byte[] u8_PopData = new Byte[s32_Count];
mi_FifoData.CopyTo(0, u8_PopData, 0, s32_Count);
mi_FifoData.RemoveRange(0, s32_Count);
return u8_PopData;
}
}
/// <summary>
/// Gets a byte without removing it from the Fifo buffer
/// returns -1 if the index is invalid
/// </summary>
public int PeekAt(int s32_Index)
{
lock (mi_FifoData)
{
if (s32_Index < 0 || s32_Index >= mi_FifoData.Count)
return -1;
return mi_FifoData[s32_Index];
}
}
}