每日速读!Pipelines

用于执行高性能的I/O，且代码不复杂

(资料图)

依赖库：System.IO.Pipelines

创建

var pipe = new Pipe();PipeReader reader = pipe.Reader;PipeWriter writer = pipe.Writer;

基本用法

// 对于socket进行读写操作async Task ProcessLinesAsync(Socket socket){// 声明一个Pipe    var pipe = new Pipe();    // 创建一个写入任务，大意是将socket中接受的数据写入到pipe中    Task writing = FillPipeAsync(socket, pipe.Writer);    // 创建一个读取任务，大意是将pipe获得数据读取出来    Task reading = ReadPipeAsync(pipe.Reader);    await Task.WhenAll(reading, writing);}// 写入任务代码async Task FillPipeAsync(Socket socket, PipeWriter writer){// 定义每个buffer为512    const int minimumBufferSize = 512;// 死循环    while (true)    {        // 获取512个byte的内存        Memory memory = writer.GetMemory(minimumBufferSize);        try        {            // 尝试从socket中读取byte数组，写入到memory中            int bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None);            //如果没有读取到数据，则不继续执行            if (bytesRead == 0)            {                break;            }            // 告知 PipeWriter 有多少数据已写入缓冲区            writer.Advance(bytesRead);        }        catch (Exception ex)        {            LogError(ex);            break;        }        // 告知PipReader，数据可使用，如果不调用则reader一直认为不可使用        FlushResult result = await writer.FlushAsync();// 写入的数据已经全部操作完成        if (result.IsCompleted)        {            break;        }    }     // 写入完成，告诉PipeReader没有多余的数据进行操作了    await writer.CompleteAsync();}// 读取写入的数据async Task ReadPipeAsync(PipeReader reader){    while (true)    {    // 读取write的数据        ReadResult result = await reader.ReadAsync();        // 获取byte数组，该数据只读        ReadOnlySequence buffer = result.Buffer;// 将result.Buffer中的数据全部提取出来，感觉类似于数据库中的游标卡尺，一步一步的读取数据        while (TryReadLine(ref buffer, out ReadOnlySequence line))        {            // Process the line.            ProcessLine(line);        }        // 告知 PipeReader 已消耗和检查了多少数据        reader.AdvanceTo(buffer.Start, buffer.End);        // 停止读取，因为在buffer中没有多余的数据了        if (result.IsCompleted)        {            break;        }    }    // 结束，告知管道可以释放内存    await reader.CompleteAsync();}// 将buffer复制到line中，里面可以自己写逻辑，例如我们的数据是以字符总数开头，以某个字符结尾等等bool TryReadLine(ref ReadOnlySequence buffer, out ReadOnlySequence line){    // 查找结尾的字符串    SequencePosition? position = buffer.PositionOf((byte)"\n");// 如果结尾字符串未找到，则返回false    if (position == null)    {        line = default;        return false;    }    // 保存要截取的数据    line = buffer.Slice(0, position.Value);    // 将剩余的数据保存起来    buffer = buffer.Slice(buffer.GetPosition(1, position.Value));    return true;}

实际中的差异

实际使用时，网络数据复制和读取后分析占用内存大小是存在差异的，因此需要根据特定环境进行配置。

假设分析数据慢于网络复制数据，则需要让网络数据多存一些然后再进行分析：

// 设置// PauseWriterThreshold：确定在调用 PipeWriter.FlushAsync 暂停之前应缓冲多少数据。// ResumeWriterThreshold：确定在恢复对 PipeWriter.FlushAsync 的调用之前，读取器必须观察多少数据。// 缓冲了10个byte后调用FlushAsync，分析了5个byte后调用PipeWriter.FlushAsyncvar options = new PipeOptions(pauseWriterThreshold: 10, resumeWriterThreshold: 5);var pipe = new Pipe(options);

PipeWriter.FlushAsync:

• 当 Pipe中的数据量超过 PauseWriterThreshold时，返回不完整的 ValueTask。
• 低于 ResumeWriterThreshold时，返回完整的 ValueTask。

使用两个值可防止快速循环，如果只使用一个值，则可能发生这种循环。

PipeScheduler

提供对异步回调运行位置的控制，默认情况下：

• 使用当前的 SynchronizationContext。
• 如果没有 SynchronizationContext，它将使用线程池运行回调。

public static void Main(string[] args){// 声明一个单线程的任务    var writeScheduler = new SingleThreadPipeScheduler();    var readScheduler = new SingleThreadPipeScheduler();    // 将声明的两个任务进行赋值    var options = new PipeOptions(readerScheduler: readScheduler,                                  writerScheduler: writeScheduler,                                  useSynchronizationContext: false);    var pipe = new Pipe(options);}public class SingleThreadPipeScheduler : PipeScheduler{    // 声明一个只读块，用于存放调用方法和读取的数据    private readonly BlockingCollection<(Action