如何使用System.IO.Pipelines和SequenceReader实现正确的ReadLine()

问题根源与修复方案 #

1. 冗余的PipeWriter.Advance调用

你在WriteAsync之后额外调用了writer.Advance(data.Length)，但WriteAsync方法本身已经会自动推进写入器的位置。这会导致管道中被添加了额外的空字节（\0），也就是你实际结果中看到的那些分隔符。

修复：删除writer.Advance(data.Length)这一行。

2. 错误的SequenceReader.TryReadTo重载

你使用了返回ReadOnlySpan<byte>的TryReadTo重载，但ReadOnlySpan只能表示连续的内存块。当行数据跨多个ReadOnlySequence段时，这个重载无法正确读取完整的行，甚至会返回错误的结果。

修复：改用返回ReadOnlySequence<byte>的重载，它可以处理非连续的序列数据，并且Encoding.UTF8.GetString支持直接将ReadOnlySequence<byte>转换为字符串。

3. AdvanceTo的正确用法

你的AdvanceTo调用其实是正确的，但需要配合修复后的读取逻辑：

• 第一个参数buffer.Start：告诉PipeReader，buffer.Start之前的数据已经处理完成，可以被回收。
• 第二个参数buffer.End：告诉PipeReader，你已经检查到buffer.End的位置，没有更多可处理的行，后续读取会等待新数据或结束信号。

修复后的完整代码 #

using System.Buffers;
using System.Diagnostics;
using System.IO.Pipelines;
using System.Text;

internal class Program
{
    public static async Task Main()
    {
        var pipe = new Pipe();
        var writer = pipe.Writer;
        var reader = pipe.Reader;

        var index = 0;
        while (true)
        {
            // 片段模拟
            if (index < Data.Length)
            {
                var data = Encoding.UTF8.GetBytes(Data[index]);
                await writer.WriteAsync(data);
                index++;
            }
            else
            {
                await pipe.Writer.CompleteAsync();
            }

            // 读取
            var result = await reader.ReadAsync();
            var buffer = result.Buffer;

            while (ReadLine(ref buffer) is { } str)
            {
                Debug.WriteLine(str);
            }

            reader.AdvanceTo(buffer.Start, buffer.End);

            if (result.IsCompleted || result.IsCanceled)
            {
                break;
            }
        }

        await reader.CompleteAsync();
    }

    private static string? ReadLine(ref ReadOnlySequence<byte> buffer)
    {
        var reader = new SequenceReader<byte>(buffer);
        if (reader.TryReadTo(out ReadOnlySequence<byte> lineSequence, "\r\n"u8))
        {
            buffer = buffer.Slice(reader.Position);
            return Encoding.UTF8.GetString(lineSequence);
        }

        return null;
    }

    private static readonly string[] Data =
    {
        "aaa",
        "bbb",
        "ccc\r\n",
        "xxx",
        "\r\n",
        "yyy",
        "\r",
        "\n"
    };
}

额外说明 #

如果需要支持单独的\n作为换行符，可以在ReadLine方法中添加对\n的检查：

private static string? ReadLine(ref ReadOnlySequence<byte> buffer)
{
    var reader = new SequenceReader<byte>(buffer);
    // 先检查\r\n
    if (reader.TryReadTo(out ReadOnlySequence<byte> lineSequence, "\r\n"u8))
    {
        buffer = buffer.Slice(reader.Position);
        return Encoding.UTF8.GetString(lineSequence);
    }
    // 再检查单独的\n
    if (reader.TryReadTo(out lineSequence, "\n"u8))
    {
        buffer = buffer.Slice(reader.Position);
        return Encoding.UTF8.GetString(lineSequence);
    }

    return null;
}

内容的提问来源于stack exchange，提问作者usausa