.Net Core中利用TPL（任务并行库）构建Pipeline处理Dataflow

在学习的过程中，看一些一线的技术文档很吃力，而且考虑到国内那些技术牛人英语都不差的，要向他们看齐，所以每天下班都在疯狂地背单词，博客有些日子没有更新了，见谅见谅

什么是TPL?

Task Parallel Library (TPL), 在.NET Framework 4微软推出TPL，并把TPL作为编写多线程和并行代码的首选方式，但是，在国内，到目前为止好像用的人并不多。（TPL）是System.Threading和System.Threading.Tasks命名空间中的一组公共类型和API 。TPL的目的是通过简化向应用程序添加并行性和并发性的过程来提高开发人员的工作效率,TPL动态地扩展并发度，以最有效地使用所有可用的处理器。通过使用TPL，您可以最大限度地提高代码的性能，让我们专注于程序本身而不用去关注负责的多线程管理。

出自： https://docs.microsoft.com/en-us/dotnet/standard/parallel-programming/task-parallel-library-tpl

为什么使用TPL?

在上面介绍了什么是TPL，可能大家还是云里雾里，不知道TPL的好处到底是什么。

我在youtube上找到了一个优秀的视频，讲述的是TPL和Thread的区别，我觉得对比一下，TPL的优势很快就能体现出来，如果大家能打开的话建议大家一定要看看。

地址是：https://www.youtube.com/watch?v=No7QqSc5cl8

现如今，我们的电脑的CPU怎么也是2核以上，下面假设我的电脑是四核的，我们来做一个实验。

使用Thread

代码中，如果使用Thread来处理任务，如果不做特出的处理，只是thread.Start(),监测电脑的核心的使用情况是下面这样的。

每一条线代表CPU某个核心的使用情况，明显，随着代码Run起来，其实只有某一个核心的使用率迅速提升，其他核心并无明显波动，为什么会这样呢？

原来，默认情况下，操作系统并不会调用所有的核心来处理任务，即使我们使用多线程，其实也是在一个核心里面运行这些Thread,而且Thread之间涉及到线程同步等问题，其实，效率也不会明显提高。

使用TPL

在代码中，引入了TPL来处理相同的任务，再次监视各个核心的使用情况，效果就变得截然不同，如下。

可以看到各个核心的使用情况都同时有了明显的提高。

说明使用TPL后，不再是使用CPU的某个核心来处理任务了，而是TPL自动把任务分摊给每个核心来处理，处理效率可想而知，理论上会有明显提升的（为什么说理论上？和使用多线程一样，各个核心之间的同步管理也是要占用一定的效率的，所以对于并不复杂的任务，使用TPL可能适得其反）。

实验结果出自https://www.youtube.com/watch?v=No7QqSc5cl8

看了这个实验讲解，是不是理解了上面所说的这句。

TPL的目的是通过简化向应用程序添加并行性和并发性的过程来提高开发人员的工作效率,TPL动态地扩展并发度，以最有效地使用所有可用的处理器。

所以说，使用TPL 来处理多线程任务可以让你不必吧把精力放在如何提高多线程处理效率上，因为这一切，TPL 能自动地帮你完成。

TPL Dataflow?

TPL处理Dataflow是TPL强大功能中的一种，它提供一套完整的数据流组件，这些数据流组件统称为TPL Dataflow Library，那么，在什么场景下适合使用TPL Dataflow Library呢？

官方举的一个 栗子 再恰当不过：

例如，通过TPL Dataflow提供的功能来转换图像,执行光线校正或防红眼，可以创建管道数据流组件,管道中的每个功能可以并行执行，并且TPL能自动控制图像流在不同线程之间的同步，不再需要Thread 中的Lock。

TPL数据流库由Block组成，Block是缓冲和处理数据的单元，TPL定义了三种最基础的Block。

source blocks（System.Threading.Tasks.Dataflow.ISourceBlock <TOutput>），源块充当数据源并且可以从中读取。

target blocks（System.Threading.Tasks.Dataflow.ITargetBlock <TInput>）,目标块充当数据接收器并可以写入。

propagator blocks（System.Threading.Tasks.Dataflow.IPropagatorBlock <TInput，TOutput>），传播器块充当源块和目标块，并且可以被读取和写入。它继承自ISourceBlock <TOutput>和ITargetBlock <TInput>。

还有其他一些个性化的Block,但其实他们都是对这三种Block进行一些扩充，可以结合下面的代码来理解这三种Block.

Code Show

1.source block 和 target block 合并成propagator block.

private IPropagatorBlock<string, Dictionary<int, string>> Process1()
        {
            var bufferBlock = new BufferBlock<Dictionary<int, string>>();
            var actionBlock = new ActionBlock<string>(x =>
              {
                  Console.WriteLine($"Process1 处理中:{x}");
                  Thread.Sleep(5000);
                  var dic = new Dictionary<int, string> { { 0, x } };
                  dic.Add(1, "Process1");
                  bufferBlock.Post(dic);
              }, new ExecutionDataflowBlockOptions
              {
                  MaxDegreeOfParallelism = _maxDegreeOfParallelism
              });
            actionBlock.Completion.ContinueWith(_ =>
            {
                Console.WriteLine($"Process1 Complete,State{_.Status}");
                bufferBlock.Complete();
            });
            return DataflowBlock.Encapsulate(actionBlock, bufferBlock);
        }

可以看到，我定义了BufferBlock和ActionBlock，它们分别继承于ISourceBlock 和 ITargetBlock ,所以说，他们其实就是源块和目标块，在new actionBlock()中传入了一个Action<String>,该Action就是该Block所执行的任务。 最后，DataflowBlock.Encapsulate(actionBlock, bufferBlock)把源块和目标块合并成了一个传递块。

2.TransformBlock

private IPropagatorBlock<Dictionary<int, string>, Dictionary<int, string>> Process2()
        {
            var block = new TransformBlock<Dictionary<int, string>, Dictionary<int, string>>(dic =>
                  {
                      Console.WriteLine($"Process2 处理中：{dic.First().Value}");
                      Thread.Sleep(5000);
                      dic.Add(2, "Process2");
                      return dic;
                  }, new ExecutionDataflowBlockOptions
                  {
                      MaxDegreeOfParallelism = _maxDegreeOfParallelism
                  }
               );

            block.Completion.ContinueWith(_ =>
            {
                Console.WriteLine($"Process2 Complete,State{_.Status}");
            });

            return block;
        }

TransfromBlock继承了IPropagatorBlock，所以它本身就是一个传递块，所以它除了要处理出入数据，还要返回数据，所以给new TransformBlock()中传入的是Func<TInput, TOutput>而不是Action<TInput>.

3.TargetBlock来收尾

private ITargetBlock<Dictionary<int, string>> Process3()
        {
            var actionBlock = new ActionBlock<Dictionary<int, string>>(dic =>
               {
                   Console.WriteLine($"Process3 处理中：{dic.First().Value}");
                   Thread.Sleep(5000);
                   dic.Add(3, "Process3");
                   Console.WriteLine("Dic中的内容如下：");
                   foreach (var item in dic)
                   {
                       Console.Write($"{item.Key}:{item.Value}||");
                   }
                   Console.WriteLine();
               }, new ExecutionDataflowBlockOptions
               {
                   MaxDegreeOfParallelism = _maxDegreeOfParallelism
               });
            return actionBlock;
        }

TargetBlock只能写入并处理数据，不能读取，所以TargetBlock适合作为Pipeline的最后一个Block。

4.控制每个Block的并行度

在在构造TargetBlock(包括其子类)的时候，可以传入ExecutionDataflowBlockOptions参数，ExecutionDataflowBlockOptions对象里面有一个MaxDegreeOfParallelism属性，通过改制，可以控制该Block的同时处理任务的数量（可以理解成线程数）。

new ExecutionDataflowBlockOptions
               {
                   MaxDegreeOfParallelism = _maxDegreeOfParallelism
               }

5.构建Pipeline,连接Block

public Task Builder()
        {
            _startBlock = Process1();
            var process2Block = Process2();
            var process3Block = Process3();

            _startBlock.LinkTo(process2Block, new DataflowLinkOptions() { PropagateCompletion = true });

            process2Block.LinkTo(process3Block, new DataflowLinkOptions() { PropagateCompletion = true });

            process3Block.Completion.ContinueWith(_ =>
            {
                Console.WriteLine($"Process3 Complete,State{_.Status}");
                Console.WriteLine("所有任务处理完成");
            });

            return process3Block.Completion;
        }

通过

ISourceBlock<TOutput>.LinkTo(ITargetBlock<TOutput> target, DataflowLinkOptions linkOption)

方法，可以把Block连接起来，即构建Pipeline，当DataflowLinkOptions对象的PropagateCompletion属性为true时，SorceBlock任务处理完成是，会把TargetBlock也标记为完成。

Block被标记为Complete 后，无法传入新的数据了，即不能再处理新的任务了。

6.Pipeline的运行

public void Process(string[] inputs)
        {
            if (inputs == null)
                return;
            foreach (var input in inputs)
            {
                _startBlock.Post(input);
            }
            _startBlock.Complete();
        }

Pipeline构建好后，我们只需要给第一个Block传入数据，该数据就会在管道内流动起来了，所有数据传入完成后，调用Block的Complete方法，把该Block标记为完成，就不可以再往里面Post数据了。

完整代码如下：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace Tpl.Dataflow
{
    public class Pipeline
    {
        IPropagatorBlock<string, Dictionary<int, string>> _startBlock;
        private int _maxDegreeOfParallelism;

        public Pipeline(int maxDegreeOfParallelism)
        {
            _maxDegreeOfParallelism = maxDegreeOfParallelism;
        }

        public void Process(string[] inputs)
        {
            if (inputs == null)
                return;
            foreach (var input in inputs)
            {
                _startBlock.Post(input);
            }
            _startBlock.Complete();
        }

        public Task Builder()
        {
            _startBlock = Process1();
            var process2Block = Process2();
            var process3Block = Process3();

            _startBlock.LinkTo(process2Block, new DataflowLinkOptions() { PropagateCompletion = true });

            process2Block.LinkTo(process3Block, new DataflowLinkOptions() { PropagateCompletion = true });

            process3Block.Completion.ContinueWith(_ =>
            {
                Console.WriteLine($"Process3 Complete,State{_.Status}");
                Console.WriteLine("所有任务处理完成");
            });

            return process3Block.Completion;
        }

        private IPropagatorBlock<string, Dictionary<int, string>> Process1()
        {
            var bufferBlock = new BufferBlock<Dictionary<int, string>>();
            var actionBlock = new ActionBlock<string>(x =>
              {
                  Console.WriteLine($"Process1 处理中:{x}");
                  Thread.Sleep(5000);
                  var dic = new Dictionary<int, string> { { 0, x } };
                  dic.Add(1, "Process1");
                  bufferBlock.Post(dic);
              }, new ExecutionDataflowBlockOptions
              {
                  MaxDegreeOfParallelism = _maxDegreeOfParallelism
              });
            actionBlock.Completion.ContinueWith(_ =>
            {
                Console.WriteLine($"Process1 Complete,State{_.Status}");
                bufferBlock.Complete();
            });
            return DataflowBlock.Encapsulate(actionBlock, bufferBlock);
        }

        private IPropagatorBlock<Dictionary<int, string>, Dictionary<int, string>> Process2()
        {
            var block = new TransformBlock<Dictionary<int, string>, Dictionary<int, string>>(dic =>
                  {
                      Console.WriteLine($"Process2 处理中：{dic.First().Value}");
                      Thread.Sleep(5000);
                      dic.Add(2, "Process2");
                      return dic;
                  }, new ExecutionDataflowBlockOptions
                  {
                      MaxDegreeOfParallelism = _maxDegreeOfParallelism
                  }
               );

            block.Completion.ContinueWith(_ =>
            {
                Console.WriteLine($"Process2 Complete,State{_.Status}");
            });

            return block;
        }

        private ITargetBlock<Dictionary<int, string>> Process3()
        {
            var actionBlock = new ActionBlock<Dictionary<int, string>>(dic =>
               {
                   Console.WriteLine($"Process3 处理中：{dic.First().Value}");
                   Thread.Sleep(5000);
                   dic.Add(3, "Process3");
                   Console.WriteLine("Dic中的内容如下：");
                   foreach (var item in dic)
                   {
                       Console.Write($"{item.Key}:{item.Value}||");
                   }
                   Console.WriteLine();
               }, new ExecutionDataflowBlockOptions
               {
                   MaxDegreeOfParallelism = _maxDegreeOfParallelism
               });
            return actionBlock;
        }
    }
}

Main方法如下：

static void Main(string[] args)
        {
            Console.WriteLine("请输入管道并发数:");
            if (int.TryParse(Console.ReadLine(), out int max))
            {
                var pipeline = new Pipeline(max);
                var task = pipeline.Builder();
                pipeline.Process(new[] { "码", "农", "阿", "宇" });
                task.Wait();
                Console.ReadKey();
            }
        }

测试运行如图：

我来解释一下，为什么是这么运行的，因为把管道的并行度设置为2，所以每个Block可以同时处理两个任务，所以,如果给管道传入四个字符 ，每个字符作为一个任务，假设传入  “码农阿宇”四个任务，会时这样的一个过程…..

  1. 码   农  两个首先进入Process1,
  2. 处理完成后，码  农   两个任务流出，
  3. Process1位置空出来， 阿  宇 两个任务流入 Process1,
  4. 码  农 两个任务流向 Process2，
  5. 阿  宇 从 Process1 处理完成后流出，此时Process1任务完成
  6. 码  农 流出 Process2 ，同时 阿 宇  流入 Process2 ……
  7. 依此类推….

该项目Github地址： https://github.com/liuzhenyulive/Tpl-Dataflow-Demo

参考文献：https://docs.microsoft.com/en-us/dotnet/standard/parallel-programming/dataflow-task-parallel-library

码字不易，如果对您有用，欢迎推荐和关注，谢谢！

原文地址：https://www.cnblogs.com/CoderAyu/p/9757389.html

时间： 2024-10-08 17:48:07

.Net Core中利用TPL（任务并行库）构建Pipeline处理Dataflow的相关文章

Asp.Net Core中利用Seq组件展示结构化日志功能

在一次.Net Core小项目的开发中,掌握的不够深入,对日志记录并没有好好利用,以至于一出现异常问题,都得跑动服务器上查看,那时一度怀疑自己肯定没学好,不然这一块日志不可能需要自己扒服务器日志来查看,果然,很多东西没掌握,至此,花点时间看了下日志的相关操作.利用日志服务来查看日志数据. 本文地址:https://www.cnblogs.com/CKExp/p/9246788.html 本文Demo的地址:https://gitee.com/530521314/LogPanel.git 一.日志

Delphi XE7中新并行库

Delphi XE7中增加了新的并行库,和.NET的Task和Parellel相似度99%. 具体内容可以看下面的文章: http://www.delphifeeds.com/go/s/119574 如果你熟悉.NET,这个功能并没有大的新意,可对Delphi程序员来说却非常实用.之前要实现多线程,唯一的办法是使用TThread类,而且过程十分繁琐.新的并行库彻底简化了这个过程. 估计下一步就该实现Async和Await异步执行结构了 另外最近Delphi的排名竟然有所上升并且进了前十,对还在用

C#5.0之后推荐使用TPL(Task Parallel Libray 任务并行库) 和PLINQ(Parallel LINQ, 并行Linq). 其次是TAP(Task-based Asynchronous Pattern, 基于任务的异步模式)

学习书籍: <C#本质论> 1--C#5.0之后推荐使用TPL(Task Parallel Libray 任务并行库) 和PLINQ(Parallel LINQ, 并行Linq). 其次是TAP(Task-based Asynchronous Pattern, 基于任务的异步模式). --用AggregateException处理Task上的未处理异常. --取消任务. CancellationToken --async修饰方法, 返回Task. task.wait(100)可以阻塞现场. a

在sharepoint中利用文档库扩展新闻应用

在sharepoint中利用文档库扩展新闻应用 本人刚刚开始使用Sharepoint,理解尚浅,错误之处请各位指正.Sharepoint的发布类模板可以建立对外的新闻类应用,但是目前在使用上还有诸多限制,不是很方便.在前一段时间给用户实施时引起了用户的很大不满.为此,特意在文档库的基础上建立了一套新闻管理系统,这套新闻系统主要有新闻发布与编辑,新闻列表.新闻浏览.分类管理等组成,主要用web part方式实现,各种参数按需传入,可以按照需要,灵活配置,下面先从与文档库结合紧密的分类管理说起:1.

.NET Core中的CSV解析库

感谢 本篇首先特别感谢从此启程兄的<.NetCore外国一些高质量博客分享>, 发现很多国外的.NET Core技术博客资源, 我会不定期从中选择一些有意思的文章翻译总结一下. .NET Core中的CSV解析库 本篇博客来源于.NET Core Totorials的<CSV Parsing In .NET Core>. 背景介绍 对于初级程序员来说, 使用string.Split(',')来解析CSV文件基本就是唯一可行的方法, 但是之后他们会发现除了使用逗号分隔值之外,CSV中

在Asp.Net Core 3.0中如何使用 Newtonsoft.Json 库序列化数据

在.Net Core 3.0中 内置了一套Json序列化/反序列化方案,默认可以不再依赖,不再支持   Newtonsoft.Json. 但是.NET Core 3.0 System.Text.Json 和 Newtonsoft.Json 使用方法不一致,对于3.0以前版本升级有限制.如果前端代码以固定更没法用了. 在Asp.Net Core 3.0中如何使用  Newtonsoft.Json 库序列化数据 官方给出了兼容处理方案,操作步骤如下: 1.引用Microsoft.AspNetCore

[翻译]在 .NET Core 中的并发编程

原文地址:http://www.dotnetcurry.com/dotnet/1360/concurrent-programming-dotnet-core 今天我们购买的每台电脑都有一个多核心的 CPU,允许它并行执行多个指令.操作系统通过将进程调度到不同的内核来发挥这个结构的优点.然而,还可以通过异步 I/O 操作和并行处理来帮助我们提高单个应用程序的性能.在.NET Core中,任务 (tasks) 是并发编程的主要抽象表述,但还有其他支撑类可以使我们的工作更容易. 并发编程 - 异步 v

在ASP.NET Core中使用Apworks开发数据服务：对HAL的支持

HAL,全称为Hypertext Application Language,它是一种简单的数据格式,它能以一种简单.统一的形式,在API中引入超链接特性,使得API的可发现性(discoverable)更强,并具有自描述的特点.使用了HAL的API会更容易地被第三方开源库所调用,并且使用起来也很方便,开发者可以像处理普通JSON数据那样去处理API数据.有关HAL的更多信息,可以参考官方网站:http://stateless.co/hal_specification.html.目前,很多REST

[译]ASP.NET Core中使用MediatR实现命令和中介者模式

作者:依乐祝 原文地址:https://www.cnblogs.com/yilezhu/p/9866068.html 在本文中,我将解释命令模式,以及如何利用基于命令模式的第三方库来实现它们,以及如何在ASP.NET Core中使用它来解决我们的问题并使代码简洁.因此,我们将通过下面的主题来进行相关的讲解. 什么是命令模式? 命令模式的简单实例以及中介者模式的简单描述 MVC中的瘦控制器是什么?我们是如如何实现使控制器变瘦的? 我们如何在我们的.NET Core应用程序中使用MediatR 使用