.NET Standard支持一组新的API，System.Span
Pipelines旨在解决.NET编写Socket通信程序时的很多困难，相信读者也对此不胜其烦，使用stream模型进行编程，就算能够解决，也是实在麻烦。
System.IO.Pipelines使用简单的内存片段来管理数据，可以极大的简化编写程序的过程。关于Pipelines的详细介绍，可以看看这里。现在ASP.NET Core中使用的Kestrel已经在使用这个API。（话说这个东西貌似就是Kestrel团队搞出来的。）
可能是直接需要用Socket场景有限（物联网用的还挺多的），Pipelines相关的资料感觉不是很多。官方给出的示例是基于ASCII协议的，有固定结尾的协议，这里我以物联网设备常用的BINARY二进制自定义协议为例，讲解基于Pipelines的程序套路。
System.IO.Pipelines
与基于Stream的方式不同，pipelines提供一个pipe，用于存储数据，pipe中间存储的数据有点链表的感觉，可以基于SequencePosition进行slice操作，这样就能得到一个ReadOnlySequence<T>对象。reader可以进行自定义操作，并在操作完成之后告诉pipe已经处理了多少数据，整个过程是不需要进行内存复制操作的，因此性能得到了提升，还少了很多麻烦。可以简单理解作为服务器端，流程：
接受数据循环：接到数据->放pipe里面->告诉pipe放了多少数据
处理数据循环：在pipe里面找一条完整数据->交给处理流程->告诉pipe处理了多少数据
协议
有一款设备，binary协议，数据包开头0x75,0xbd,0x7e,0x97一共4个字节，随后跟数据包长度2个字节（固定2400字节，不固定长度也可以参照），随后是数据区。在设备连接成功之后，数据主动从设备发送到PC。
关键代码
虽然是.NET Core平台的，但是.NET FRAMEWORK 4.6.1上面也可以nuget安装，直接
install-package system.io.pipelines
进行安装就可以了。Socket相关处理的代码不再写了，只列关键的。
代码第一步是声明pipe。
private async void InitPipe(Socket socket)
{
Pipe pipe = new Pipe();
Task writing = FillPipeAsync(socket,pipe.Writer);
Task reading = ReadPipeAsync(socket,pipe.Reader);
await Task.WhenAll(reading,writing);
}
pipe有reader还有一个writer，reader负责读取pipe数据，主要用在数据处理循环，writer负责将数据写入pipe，主要用在数据接受循环。
//写入循环
private async Task FillPipeAsync(Socket socket,PipeWriter writer)
{
//数据流量比较大，用1M字节作为buffer
const int minimumBufferSize = 1024 * 1024;
while (running)
{
try
{
//从writer中，获得一段不少于指定大小的内存空间
Memory<byte> memory = writer.GetMemory(minimumBufferSize);
//将内存空间变成ArraySegment，提供给socket使用
if (!MemoryMarshal.TryGetArray((ReadOnlyMemory<byte>)memory,out ArraySegment<byte> arraySegment))
{
throw new InvalidOperationException("Buffer backed by array was expected");
}
//接受数据
int bytesRead = await SocketTaskExtensions.ReceiveAsync(socket,arraySegment,SocketFlags.None);
if (bytesRead == 0)
{
break;
}
//一次接受完毕，数据已经在pipe中，告诉pipe已经给它写了多少数据。
writer.Advance(bytesRead);
}
catch
{
break;
}
// 提示reader可以进行读取数据，reader可以继续执行readAsync()方法
FlushResult result = await writer.FlushAsync();
if (result.IsCompleted)
{
break;
}
}
// 告诉pipe完事了
writer.Complete();
}
//读取循环
private async Task ReadPipeAsync(Socket socket,PipeReader reader)
{
while (running)
{
//等待writer写数据
ReadResult result = await reader.ReadAsync();
//获得内存区域
ReadOnlySequence<byte> buffer = result.Buffer;
SequencePosition? position = null;
do
{
//寻找head的第一个字节所在的位置
position = buffer.PositionOf((byte)0x75);
if (position != null)
{
//由于是连续四个字节作为head，需要进行比对，我这里直接使用了ToArray方法，还是有了内存拷贝动作，不是很理想，但是写起来很方便。
//对性能有更高要求的场景，可以进行slice操作后的单独比对，这样不需要内存拷贝动作
var headtoCheck = buffer.Slice(position.Value,4).ToArray();
//SequenceEqual需要引用System.Linq
if (headtoCheck.SequenceEqual(new byte[] { 0x75,0x97 }))
{
//到这里，认为找到包开头了（从position.value开始），接下来需要从开头处截取整包的长度，需要先判断长度是否足够
if (buffer.Slice(position.Value).Length >= 2400)
{
//长度足够，那么取出ReadOnlySequence，进行操作
var mes = buffer.Slice(position.Value,2400);
//这里是数据处理的函数，可以参考官方文档对ReadOnlySequence进行操作，文档里面使用了span，那样性能会好一些。我这里简单实用ToArray()操作，这样也有了内存拷贝的问题，但是处理的直接是byte数组了。
await ProcessMessage(mes.ToArray());
//这一段就算是完成了，从开头位置，一整个包的长度就算完成了
var next = buffer.GetPosition(2400,position.Value);
//将buffer处理过的舍弃，替换为剩余的buffer引用
buffer = buffer.Slice(next);
}
else
{
//长度不够，说明数据包不完整，等下一波数据进来再拼接，跳出循环。
break;
}
}
else
{
//第一个是0x75但是后面不匹配，可能有数据传输问题，那么需要舍弃第一个，0x75后面的字节开始再重新找0x75
var next = buffer.GetPosition(1,position.Value);
buffer = buffer.Slice(next);
}
}
}
while (position != null);
//数据处理完毕，告诉pipe还剩下多少数据没有处理（数据包不完整的数据，找不到head）
reader.AdvanceTo(buffer.Start,buffer.End);
if (result.IsCompleted)
{
break;
}
}
reader.Complete();
}
以上代码基本解决了以下问题：
- 数据接收不完整，找不到开头结尾，导致数据大量丢弃，或者自己维护一个queue的代码复杂性
- 数据接收与处理的同步问题
- 一次性收到多条数据的情况
后记
本文只是解释了pipeline处理的模式，对于茫茫多的ToArray方法，可以使用基于Span的操作进行优化（有时间就来填坑）。另外，如果在await ProcessMessage(mes.ToArray());这里，直接使用Task.Run(()=>ProcessMessage(mes);代替的话，实测会出现莫名其妙的问题，很有可能是pipe运行快，在系统调度Task之前，已经将内存释放导致的，如果需要优化这一块的话，需要格外注意。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。