本文主要是介绍C# 初识System.IO.Pipelines,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
写在前面
在进一步了解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));
}
调用示例
总结
例子中用到的文本文件是一个以\n 换行符作为结尾的多行文本,微软官方示例没有提供,这个是自己建的测试文件,如果没有检测到\n会抛出异常。
从运行的结果可以看到,从传入的流中识别以\n结尾,作为数据块的区分,利用这个特性定义数据报文的尾部,实现分包。
这篇关于C# 初识System.IO.Pipelines的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!