并发系列64章(TPL 数据流)第七章
时空道人 人气:0前言
什么是TPL?全称:transmission control protocol
传输层对应于OSI七层参考模型的传输层,它提供两种端到端的通信服务。
然后思维方式回到为什么有这个TPL 数据流上。
TPL 数据流库向具有高吞吐量和低滞后时间的占用大量 CPU 和 I/O 操作的应用程序的并行化和消息传递提供了基础。 它还能显式控制缓存数据的方式以及在系统中移动的方式。
为了更好地了解数据流编程模型,请考虑一个以异步方式从磁盘加载图像并创建复合图像的应用程序。
传统编程模型通常需要使用回调和同步对象(例如锁)来协调任务和访问共享数据。
通过使用数据流编程模型,您可以从磁盘读取时创建处理图像的数据流对象。
在数据流模型下,您可以声明当数据可用时的处理方式,以及数据之间的所有依赖项。 由于运行时管理数据之间的依赖项,因此通常可以避免这种要求来同步访问共享数据。
此外,因为运行时计划基于数据的异步到达,所以数据流可以通过有效管理基础线程提高响应能力和吞吐量。
分析一下,这段话。
TPL 数据流库向具有高吞吐量和低滞后时间的占用大量 CPU 和 I/O 操作的应用程序的并行化和消息传递提供了基础。
解决一个问题就是:
高吞吐量和低滞后时间的占用大量 CPU 和 I/O 操作的应用程序。
如何解决的:
应用程序的并行化和消息传递提供了基础。通过并行解决的。
例子:
异步方式从磁盘加载图像并创建复合图像的应用程序
遇到的问题:
协调任务和访问共享数据 需要 回调和同步。
就是说共享数据的时候,需要同步。
总结问题:共享数据代价大。
如果解决的:
由于运行时管理数据之间的依赖项,因此通常可以避免这种要求来同步访问共享数据。
总结:解决了依赖,那么不需要同步了。
综上所述:TPL 数据流库的作用在于解决数据之间的依赖,避免同步访问共享数据。
正文
链接数据流块
var multiplyBlock = new TransformBlock<int, int>(item=>item*2);
var subtractBlock = new TransformBlock<int, int>(item=> { Console.WriteLine(item); return item - 2; });
multiplyBlock.LinkTo(subtractBlock);
multiplyBlock.Post(10);
Console.ReadKey();
打印出来就是20了。
传递出错信息
需要处理数据流网格中发生的错误
传递出错信息
如果数据流块的委托抛出异常,这个块就是故障块。一但数据流进入了故障状态,就会删除所有数据(停止接收新的数据)。
什么意思呢?
static async void datalfow()
{
var multiplyBlock = new TransformBlock<int, int>(item =>
{
if (item == 1)
{
throw new InvalidOperationException("not good");
}
return item * 2;
}
);
var subtractBlock = new TransformBlock<int, int>(item => { Console.WriteLine(item); return item - 2; });
multiplyBlock.LinkTo(subtractBlock, new DataflowLinkOptions { PropagateCompletion = true});
try
{
multiplyBlock.Post(10);
multiplyBlock.Post(1);
multiplyBlock.Post(20);
await subtractBlock.Completion;
}
catch(AggregateException e)
{
Console.WriteLine(e);
}
}
结果是:
有没有发现multiplyBlock.Post(20);,没有运行?
因为一但一个有错误,那么就会终止,并销毁数据。
这里和上面不同的是,new DataflowLinkOptions { PropagateCompletion = true}。
多个这个东西,那么这个有啥用呢?
因为我们链接块的时候,这个库不会帮助我们传递块运行的状态,如果不传递的话,下一个块是不晓得上一个块到底啥情况,这样不利于我们捕获异常。
而这种传递做法,我们只要在最后的处理模块,统一处理错误就可以。
断开链接
这个我从来就没有遇到过。是这样子的,适用一种这样的场景。
比如说有一个数据块需要动态替换,需要断开现有的模块然后接上新的数据块。
static async void datalfow()
{
var multiplyBlock = new TransformBlock<int, int>(item =>
{
if (item == 1)
{
throw new InvalidOperationException("not good");
}
Console.WriteLine("item:" + item);
return item * 2;
}
);
var subtractBlock = new TransformBlock<int, int>(item => { Console.WriteLine(item-2); return item - 2; });
var appendBlock = new TransformBlock<int, int>(item => { Console.WriteLine(item+2); return item + 2; });
var link=multiplyBlock.LinkTo(subtractBlock, new DataflowLinkOptions { PropagateCompletion = true});
try
{
for (int i = 0; i < 20; i++)
{
multiplyBlock.Post(2);
if (i==10)
{
await Task.Delay(1000);
link.Dispose();
multiplyBlock.LinkTo(appendBlock, new DataflowLinkOptions { PropagateCompletion = true });
}
}
await subtractBlock.Completion;
}
catch(AggregateException e)
{
Console.WriteLine(e);
}
}
结果是:
值得注意的是,我这里了一个:
await Task.Delay(1000);
这是模拟动态运行的时候,因为当我post结束的时候,数据块链接还没开始传递。
注:
除非保证链接是空闲的情况下,否则在断开数据块的链接时候会出现竞争。
竞争的是先断开还是先传递。
但是这种竞争是安全的,他会保证要不断开,要不传递带下一个数据块。
限制流量
前面我们都是线性链接,就是一条路走到黑。但是呢,有时候出现分叉的时候,那么该如何均衡呢?
之所以考虑均衡,是因为比如传递到下一个数据块的时候,是会有缓存的。如果有条分叉,一条分叉无限去缓存,那另外一条可能吃不上饭了。
static async void datalfow()
{
var multiplyBlock = new TransformBlock<int, int>(item =>
{
if (item == 1)
{
throw new InvalidOperationException("not good");
}
Console.WriteLine("item:" + item);
return item * 2;
}
);
var subtractBlock = new TransformBlock<int, int>(item => { Console.WriteLine(item-2); return item - 2; });
var appendBlock = new TransformBlock<int, int>(item => { Console.WriteLine(item+2); return item + 2; });
multiplyBlock.LinkTo(subtractBlock, new DataflowLinkOptions { PropagateCompletion = true});
multiplyBlock.LinkTo(appendBlock, new DataflowLinkOptions { PropagateCompletion = true });
try
{
for (int i = 0; i < 100; i++)
{
multiplyBlock.Post(2);
}
await subtractBlock.Completion;
}
catch(AggregateException e)
{
Console.WriteLine(e);
}
}
这种就属于没吃上饭的情况。
static async void datalfow()
{
var multiplyBlock = new TransformBlock<int, int>(item =>
{
if (item == 1)
{
throw new InvalidOperationException("not good");
}
Console.WriteLine("item:" + item);
return item * 2;
}
);
var options = new DataflowBlockOptions {BoundedCapacity=1 };
var subtractBlock = new TransformBlock<int, int>(item => {
return item - 2;
}, options);
var appendBlock = new TransformBlock<int, int>(item => { Console.WriteLine(item+2); return item + 2; }, options);
multiplyBlock.LinkTo(subtractBlock, new DataflowLinkOptions { PropagateCompletion = true});
multiplyBlock.LinkTo(appendBlock, new DataflowLinkOptions { PropagateCompletion = true });
try
{
for (int i = 0; i < 100; i++)
{
multiplyBlock.Post(2);
}
await subtractBlock.Completion;
}
catch(AggregateException e)
{
Console.WriteLine(e);
}
}
限制缓存为1,那么这时候我们就会相互切换。
下一章
整理:
1.数据流块的并行处理
2.创建自定义数据流块
参考
https://www.cnblogs.com/yswenli/p/8042594.html
加载全部内容