在C#中缓冲字节数据

Question

我的应用程序从TCP套接字读取字节并需要缓冲它们,以便我可以稍后从中提取消息.由于TCP的性质,我可能在一次读取中得到部分或多个消息,因此在每次读取之后,我想检查缓冲区并提取尽可能多的完整消息.

因此,我想要一个允许我执行以下操作的课程:

• 将任意byte []数据附加到它
• 在不消费内容的情况下检查内容,特别是检查内容量并搜索某个字节或字节的存在
• 以byte []的形式提取和使用部分数据,而将其余部分留在那里以供将来读取

我希望我想要的可以用.NET库中的一个或多个现有类来完成,但我不确定哪些类.System.IO.MemoryStream看起来接近我想要的,但是(a)不清楚它是否适合用作缓冲区(读取数据是否从容量中删除？)和(b)读写似乎发生在同一个地方 - "流的当前位置是下一次读取或写入操作可能发生的位置." - 这不是我想要的.我需要写到最后并从前面阅读.

Answer 1

我建议你MemoryStream在引擎盖下使用,但将其封装在另一个存储的类中:

• 该 MemoryStream
• 目前的"阅读"位置
• 目前"消费"的位置

然后它会暴露:

• 写入:将流的位置设置为结尾,写入数据,将流的位置设置回读取位置
• 读取:读取数据,将读取位置设置为流的位置
• 消耗:更新消耗的位置(根据您尝试消费的方式详细说明); 如果消耗位置高于某个阈值,则将现有的缓冲数据复制到新的MemoryStream并更新所有变量.(您可能不希望在每个使用请求上复制缓冲区.)

请注意,如果没有额外的同步,这些都不是线程安全的.

Answer 2

只需使用一个大字节数组和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];
        }
    }
}

请注意:代码"消耗"读取 - 如果您不希望只删除"_startIndex = ..."部分(或使重载可选参数和检查或其他).