当前位置：首页 > news >正文

C# 初识System.IO.Pipelines

news 2025/9/17 10:31:38

写在前面

在进一步了解Socket粘包分包的过程中，了解到了.NET 中的 System.IO.Pipelines，可以更优雅高效的解决这个问题；先跟随官方的示例做个初步的认识。

System.IO.Pipelines 是一个库，旨在使在 .NET 中执行高性能 I/O 更加容易。该库的目标为适用于所有 .NET 实现的 .NET Standard。
System.IO.Pipelines 具有高性能的流数据分析功能，可以减少代码复杂性。

老规矩通过NuGet安装该类库

代码实现

using System.Buffers;
using System.IO.Pipelines;
using System.Text;class Program
{static async Task Main(){using var stream = File.OpenRead("lorem-ipsum.txt");var reader = PipeReader.Create(stream);var writer = PipeWriter.Create(Console.OpenStandardOutput(),new StreamPipeWriterOptions(leaveOpen: true));WriteUserCancellationPrompt();var processMessagesTask = ProcessMessagesAsync(reader, writer);var userCanceled = false;var cancelProcessingTask = Task.Run(() =>{while (char.ToUpperInvariant(Console.ReadKey().KeyChar) != 'C'){WriteUserCancellationPrompt();}userCanceled = true;// No exceptions thrownreader.CancelPendingRead();writer.CancelPendingFlush();});await Task.WhenAny(cancelProcessingTask, processMessagesTask);Console.WriteLine($"\n\nProcessing {(userCanceled ? "cancelled" : "completed")}.\n");}static void WriteUserCancellationPrompt() =>Console.WriteLine("Press 'C' to cancel processing...\n");static async Task ProcessMessagesAsync(PipeReader reader,PipeWriter writer){try{while (true){ReadResult readResult = await reader.ReadAsync();ReadOnlySequence<byte> buffer = readResult.Buffer;try{if (readResult.IsCanceled){break;}if (TryParseLines(ref buffer, out string message)){FlushResult flushResult =await WriteMessagesAsync(writer, message);if (flushResult.IsCanceled || flushResult.IsCompleted){break;}}if (readResult.IsCompleted){if (!buffer.IsEmpty){throw new InvalidDataException("Incomplete message.");}break;}}finally{reader.AdvanceTo(buffer.Start, buffer.End);}}}catch (Exception ex){Console.Error.WriteLine(ex);}finally{await reader.CompleteAsync();await writer.CompleteAsync();}}static bool TryParseLines(ref ReadOnlySequence<byte> buffer,out string message){SequencePosition? position;StringBuilder outputMessage = new();while (true){position = buffer.PositionOf((byte)'\n');if (!position.HasValue)break;outputMessage.Append(Encoding.ASCII.GetString(buffer.Slice(buffer.Start, position.Value))).AppendLine();buffer = buffer.Slice(buffer.GetPosition(1, position.Value));};message = outputMessage.ToString();return message.Length != 0;}static ValueTask<FlushResult> WriteMessagesAsync(PipeWriter writer,string message) =>writer.WriteAsync(Encoding.ASCII.GetBytes(message));
}