C# 并发编程学习笔记

标签：Task item C# 编程 var 并发数据流 new 数据

术语含义

并发：同时做多件事情；

多线程：并发的一种形式，它采用多个线程来执行程序；

并行处理：把正在执行的大量任务分割成小块，分配给多个同时运行的线程；

异步编程：并发的一种形式，采用future模式或回调（callback）机制，以避免产生不必要的线程。启动了的操作将会在一段时间后完成；

响应式编程：一种声明式的编程模式，程序在该模式中对事件作出响应；

使用new Thread表明项目代码太过时了；

并行的形式：

数据并行，指有大量的数据需要处理，并且每一块数据的处理过程基本上是彼此独立的；

任务并行，指需要执行大量任务，并且每个任务的执行过程基本上是彼此独立的；

数据并行的方法：

1.Parallel.Foreach，对资源更友好

2.PLINQ，values.AsParallel().Select(val=>IsPrime(val));

-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

异步编程

1.报告进度

使用IProgress<T>和Progress<T>类型

var progress = new Progress<double>();
progress.ProgressChanged += (sender,args) => { ... };
await MyMethodAsync(progress);

调用
if(progress!=null)
    progress.Report(...);

Progress<T>在创建时会捕获当前上下文，并且在这个上下文中调用回调函数。

所以如果UI线程中创建了Progress<T>，就能在回调函数中更新UI，即使异步方法是在后台线程中调用Report的。

2.等待一组任务完成

正常await Task.WhenAll只捕获一个异常，如果需要捕获WhenAll中的所有异常，需要使用 AggregateException allExceptions = allTasks.Exception

3.等待任意一个任务完成

Task.WhenAny返回的task对象永远不会以“故障”或“已取消”状态作为结束，如果这个任务完成时有异常，异常也不会传递给Task.WhenAny返回的Task对象。因此，通常需要在Task对象完成后继续使用await；

4.按顺序处理已完成的任务

static async Task<int> DelayAndReturnAsync(int val)
{
    await Task.Delay(TimeSpan.FromSeconds(val));
    return val;
}

static async Task ProcessTasksAsync()
{
    Task<int> taskA = DelayAndReturnAsync(2);
    Task<int> taskB = DelayAndReturnAsync(3);
    Task<int> taskC = DelayAndReturnAsync(1);
    var tasks = new[] { taskA, taskB, taskC };

    // 输出2，3，1
    //foreach(var task in tasks)
    //{
    //    var result = await task;
    //    Console.WriteLine(result);
    //}

    // 对序列求值，创建集合 = 开始执行任务
    // 输出1，2，3
    var processingTasks = tasks.Select(async t =>
    {
        var result = await t;
        Console.WriteLine(result);
    }).ToArray();

    await Task.WhenAll(processingTasks);

    // 使用NuGet包 Nito.Async
    //foreach (var task in tasks.OrderByCompletion())
    //{
    //    var result = await task;
    //    Console.WriteLine(result);
    //}
}

5.避免上下文延续

如果大量async方法在UI上下文中恢复，可能会引起性能上的问题，如果没有必要恢复到原来的上下文，就用ConfigureAwait，如果一个async方法的一部分需要上下文，一部分不需要上下文，则可以考虑把它拆分为两个（或更多）async方法。微软公布了一个指导标准：每秒在UI线程中有100个左右的延续任务还尚可，每秒1000个就太多了。

6.处理async Task方法异常

在async Task方法中引发的异常，存放在返回的Task对象中，只有当Task对象被await调用时，才会引发异常。

7.处理async void方法异常

async void方法没有返回Task对象，无法存放异常，故无法在try..catch中捕获void方法中抛出的异常；

解决方法一：可以重载一个返回Task类型的重载，这样就能在await中捕获异常

解决方法二：使用NuGet包Nito.AsyncEx

try
{
    AsyncContext.Run(() => objectUnderTest.ExecuteWithoutTask("some argument"));
}
catch (Exception ex)
{
    exception = ex;
}

并行编程

用于分解计算密集型的任务片段，如果是I/O密集型任务请用异步编程方法。

Parallel类和并行LINQ只是为了使用方便，从而对Task类进行了封装。

1.数据并行处理

ParallelLoopResult parallelLoopResult =
    Parallel.ForEach(
    matrices, //index为迭代次数，state是循环状态
        (matrix, state, index) =>
        {
            if (!matrix.IsInvertible)
            {
                state.Stop(); //执行Stop之后ParallelLoopResult.IsCompleted=false
            }
            else
            {
                matrix.Invert();
            }

return parallelLoopResult;

更常见的做法是取消并行循环（外部cancel），传入一个CancellationTokenSource

var cts = new CancellationTokenSource();

ParallelLoopResult parallelLoopResult = Parallel.ForEach( matrices, new ParallelOptions { CancellationToken = token }, (matrix, state, index) => {//....});

cts.Cancel();//如果并行循环在运行时会抛出异常

Parallel.For方法多用于多个数组的数据共用相同的索引的情况。

2.并行聚合（累加和，平均值等）

Parallel类通过local value概念实现聚合，使用LocalFinally委托对每个局部值进行聚合，LocalFinally委托执行的线程跟Body是同一个线程。

Parallel.ForEach(
    source: values,
    localInit: () => 0,
    body: (item, state, localValue) => localValue + item,
    localFinally: (localValue) =>
    {
        // localFinally is invoke for each working task.
        lock (mutext)
        {
        // 以同步的方式对存放结果的变量进行访问
            result += localValue;
        }
    });

使用PLINQ对聚合的支持更有表现力，代码也更简洁

values.AsParallel().Sum();

// 使用累加器函数
values.AsParallel().Aggregate(seed: 0, func: (sum, item) => sum + item);

3.并行调用

使用场景：需要并行调用一批方法，并且这些方法大部分都是相互独立的。

Parallel.Invoke(
                () => ProcessPartialArray(array, 0, array.Length / 2),
                () => ProcessPartialArray(array, array.Length / 2, array.Length));

对于简单的并行调用，Parallel.Invoke是不错的解决方案。如果要对每一个输入的数据调用一个操作（改用Foreach），或者每一个操作产生了一些输出（改用并行LINQ）；

4.动态并行

使用场景：并行任务的结构和数量要在运行时才能确定

private static void ProcessTree(Node root)
{
    var task = Task.Factory.StartNew(() => Traverse(root), CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
    task.Wait();
}

private static void Traverse(Node current)
{
    Console.WriteLine(string.Format("Current Node：{0}\t Task：{1}\t Thread：{2}", current.Name, Task.CurrentId, Thread.CurrentThread.ManagedThreadId));
    //Thread.Sleep(500);
    if (current.Left != null)
    {
        //TaskCreationOptions.AttachedToParent将父任务与子任务同步。 默认情况下，子任务（即由外部任务创建的内部任务）将独立于其父任务执行。
        Task.Factory.StartNew(() => Traverse(current.Left), CancellationToken.None,
            TaskCreationOptions.AttachedToParent, TaskScheduler.Default);
    }
    if (current.Right != null)
    {
        Task.Factory.StartNew(() => Traverse(current.Right), CancellationToken.None,
            TaskCreationOptions.AttachedToParent, TaskScheduler.Default);
    }
}

如果任务没有“父/子”关系，那可以使用任务延续(continuation)的方法，安排任务一个接一个地运行。

在并发编程中，Task有两个作用：作为并行任务，或作为异步任务。

并行任务可以使用阻塞的成员函数，例如Task.Wait、Task.Result、Task.WaitAll等等，通常也是用AttachedToParent建立任务之间的“父/子”关系。并行任务的创建需要用Task.Run或者Task.Factory.StartNew。

相反，异步任务应该避免使用阻塞的成员函数，而应该使用await、Task.WhenAll和Task.WhenAny，异步任务不适用AttachedToParent，但可以通过await另一个任务，建立一种隐式的“父/子”关系。

5.并行LINQ

使用场景：需要对一批数据进行并行处理，生成另外一批数据，或者对数据进行统计。

PLINQ为各种各样的操作提供了并行版本，包括where，select以及各种聚会运算，例如sum、average和更通用的aggregate。

数据流基础

TPL数据流可用来创建网格（mesh）和管道（pipleline），并通过它们以异步方式发送数据，这种“声明式编程”风格通常要先完整地定义网格，然后才能开始处理数据。

每个网格由各种相互链接的数据流块（block）构成，每个块只负责处理某个单独的步骤。

使用TPL数据流需要安装NuGet包：Microsoft.Tpl.Dataflow

BufferBlock：最基础的Block，提供FIFO处理

ActionBlock：顺序处理委托；

TransformBlock：充当数据转换处理的功能功能，内部维护了2个Queue，一个InputQueue，一个OutputQueue。InputQueue存储输入的数据，而通过Transform处理以后的数据则放在OutputQueue，通过Receive方法来阻塞的一个一个获取OutputQueue中的数据，只有OutputQueue中有数据时才会返回；

TransformManyBlock：一个输入数据可以对应多个输出数据；

BroadcastBlock：使所有与它链接的目标块都收到数据的副本，BroadcastBlock不保存数据，每一个数据发送到所有接收者之后，这条数据就会被后面最新的数据所覆盖，如果没有链接目标块，数据将会被丢弃，但BroadcastBlock总是会保存最后一个数据，如果有一个新的链接目标块，那么这个目标块就会收到这个数据；

WriteOnceBlock：最最简单的Block，它只能储存一个数据，一旦这个数据被发送出去以后，这个数据还是留在Block中，并且不会被删除或被替换；

BatchBlock：提供把多个单个的数据组合起来处理的功能，在数据量满足构造函数中定义的要求时，会处理放进OutputQueue中，当BatchBlock调用Complete告知Post数据结束的时候，会把InputQueue中剩余的数据打包放入OutputQueue中等待处理，而不管数量是否满足构造函数的要求。

BatchBlock两种执行模式：

//Greedy=false 非贪婪模式 需从N个Source接收一个数据后进行处理，Source指LinkTo连接到这个BatchBlock的Block，这种时候需要Source.Complete()并且等待Completion完成；
//Greedy=true 贪婪模式（默认） 从任何Post到batchBlock接收到目标个数后进行处理。

JoinBlock：等待一个数据组合，组合作为一个Tuple传递给目标Block，如果定义了JoinBlock<int, string>类型，那么JoinBlock内部会有两个ITargetBlock，一个接收int类型的数据，一个接收string类型的数据。只有两个Block都收到数据后，才会放到JoinBlock的OutputQueue中输出。

BatchedJoinBlock：BacthBlock和JoinBlick的组合。JoinBlock是组合目标队列的一个数据，而BatchedJoinBlock是组合目标队列的N个数据，当然这个N可以在构造函数中配置。只要2个ITargetBlock中的数据个数加起来等于N就可以输出到OutputQueue。

1.链接数据流块

var multiplyBlock = new TransformBlock<int, int>(item => item * 2);
var subtractBlock = new TransformBlock<int, int>(item => item - 2);
multiplyBlock.LinkTo(subtractBlock);
// 默认情况下，链接的数据流块只传递数据，不传递完成情况（或出错信息），要实现完成情况的传递，需要设置PropagateCompletion属性
multiplyBlock.LinkTo(subtractBlock, new DataflowLinkOptions { PropagateCompletion = true });

// 向Dataflow发送信号,停止接收数据
multiplyBlock.Complete();
await multiplyBlock.Completion;

Completion属性只有在设置了Complete方法后才会有效

2.处理数据流网格中发生的错误

// 如果数据流块内的委托抛出错误，这个块就进入故障状态，一旦数据流块进入故障状态，就会删除所有的数据（并停止接收新数据）
static async Task TransferException()
{
    try
    {
        var block = new TransformBlock<int, int>(item =>
        {
            if (item == 1)
                throw new InvalidOperationException("DataflowError");
            return item * 2;
        });
        block.Post(1);
        block.Post(2);
        block.Complete();
        // Completion属性返回一个任务，一旦数据流块执行完成，这个任务也完成，如果数据流块出错，这个任务也出错
        await block.Completion;

    }
    catch (InvalidOperationException) { }
}

如果用PropagateCompletion这个参数传递完成情况，错误信息也会被传递，只不过被封装在AggregateException类中传递给下一个块。

数据流块收到传过来的出错信息后，即使它已经被封装在AggregateException中，仍会用AggregateException进行封装，经过多个链接后才被发现，这个原始错误就会被AggregateException封装很多层，这时候用AggregateException.Flatten可以简化错误处理过程。

3.断开链接

链接建立后是无法修改过滤器的，必须先断开链接，然后用新的过滤器建立链接（DataflowLinkOptions.Append=false代表将链接前置）

IDisposable link = multiplyBlock.LinkTo(subtractBlock);
//....
link.Dispose();

multiplyBlock.LinkTo(subtractBlock, new DataflowLinkOptions { Append = false });

4.限制流量

默认情况下，数据流块生成输出的数据后，会按照创建的次序逐个地尝试通过链接传递数据，每个数据流块维护一个输入缓冲区，在处理数据之前接收任意数量的数据。

有分叉是，一个源块链接了两个目标块，上面的做法就会导致第一个目标块不停地缓冲数据（目标块还来不及处理数据时就得对所有数据进行缓冲），第二个目标块永远没有机会得到数据。

解决方法是使用BoundedCapacity属性，来限制目标块的流量。

var sourceBlock = new BufferBlock<int>();
var options = new DataflowBlockOptions { BoundedCapacity = 1 };
var tagetBlockA = new BufferBlock<int>(options);
var tagetBlockB = new BufferBlock<int>(options);
sourceBlock.LinkTo(tagetBlockA);
sourceBlock.LinkTo(tagetBlockB);

5.数据流块的并行处理

默认情况下每个数据流块是互相独立的，即使将它们链接起来后，它们也是独立运行的，因此每个数据流网格本身就有并行特性。

假如某个特定的数据流块的计算量特别大，那就可以设置MaxDegreeOfParallelism参数，使数据流块在处理输入的数据时采用并行的方式，默认值是1。

var multiplyBlock = new TransformBlock<int, int>(item => item * 2, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded });

难点：如何找出需要并行处理的数据流块？

解决方法：在调试时暂停数据流运行，在调试器中查看等待的数据线的数量（就是还没有被数据流块处理的数据项），如果等待的数据项很多就表明需要进行并行化处理。

6.创建自定义的数据流块

希望一些可重用的程序逻辑在自定义数据流块中使用，可以通过使用Encapsulate方法，取出数据流网格中任何具有单一输入块和输出块的部分。

IPropagatorBlock<int, int> CreateMyCustomBlock()
{
    var multiplyBlock = new TransformBlock<int, int>(item => item * 2);
    var addBlcok = new TransformBlock<int, int>(item => item + 2);
    var divideBlock = new TransformBlock<int, int>(item => item / 2);
    var flowCompletion = new DataflowLinkOptions { PropagateCompletion = true };
    multiplyBlock.LinkTo(addBlcok, flowCompletion);
    addBlcok.LinkTo(divideBlock, flowCompletion);
    return DataflowBlock.Encapsulate(multiplyBlock, divideBlock);
}

标签：Task,item,C#,编程,var,并发,数据流,new,数据
From： https://www.cnblogs.com/zhengzc/p/13745322.html

术语含义

异步编程

1.报告进度

2.等待一组任务完成

3.等待任意一个任务完成

4.按顺序处理已完成的任务

5.避免上下文延续

6.处理async Task方法异常

7.处理async void方法异常

并行编程

1.数据并行处理

2.并行聚合（累加和，平均值等）

3.并行调用

4.动态并行

在并发编程中，Task有两个作用：作为并行任务，或作为异步任务。

5.并行LINQ

数据流基础

1.链接数据流块

2.处理数据流网格中发生的错误

3.断开链接

4.限制流量

5.数据流块的并行处理

6.创建自定义的数据流块

相关文章

赞助商

阅读排行