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

时间 2019-11-13

标签 core 利用 tpl 任务并行构建 pipeline 处理 dataflow 栏目设计模式繁體版

原文原文链接

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

什么是TPL?

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

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

为何使用TPL?

在上面介绍了什么是TPL，可能你们仍是云里雾里，不知道TPL的好处究竟是什么。多线程

我在youtube上找到了一个优秀的视频，讲述的是TPL和Thread的区别，我以为对比一下，TPL的优点很快就能体现出来，若是你们能打开的话建议你们必定要看看。并发

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

现现在，咱们的电脑的CPU怎么也是2核以上，下面假设个人电脑是四核的，咱们来作一个实验。学习

使用Thread

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

每一条线表明CPU某个核心的使用状况，明显，随着代码Run起来，其实只有某一个核心的使用率迅速提高，其余核心并没有明显波动，为何会这样呢？ui

原来，默认状况下，操做系统并不会调用全部的核心来处理任务，即便咱们使用多线程，其实也是在一个核内心面运行这些Thread,并且Thread之间涉及到线程同步等问题，其实，效率也不会明显提升。spa

使用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能够同时处理两个任务，因此,若是给管道传入四个字符，每一个字符做为一个任务，假设传入 “码农阿宇”四个任务，会时这样的一个过程…..

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

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

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

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