Parallel 与 ConcurrentBag＜T＞这对儿黄金搭档（C#）【并发编程系列】

ゞ浴缸里的玫瑰 2024-03-17 20:42 9阅读 0赞

### 〇、前言 ###

日常开发中经常会遇到数据统计，特别是关于报表的项目。数据处理的效率和准确度当然是首要关注点。

本文主要介绍，如何通过 Parallel 来并行处理数据，并组合 ConcurrentBag<T> 集合，来将处理效率达到高点的同时，也能确保数据的准确。

[回到顶部][Link 1]

### 一、ConcurrentBag<T> 简介 ###

#### 1、简介与源码 ####

ConcurrentBag<T>，表示对象的线程安全的无序集合。ConcurrentBag 内部将数据按线程的标识独立进行存储，程序可以在同一个线程中插入、删除元素，所以每个线程对其数据的操作是非常快的。

下面是源码供参考：

点击展开 ConcurrentBag 源码

#### 2、属性 ####

**Count**

获取 ConcurrentBag<T> 中包含的元素数

**IsEmpty**

获取一个值，该值指示 ConcurrentBag<T> 是否为空

#### 3、方法 ####

**Add(T)**

将对象添加到 ConcurrentBag<T> 中。

**Clear()**

从 ConcurrentBag<T> 中删除所有值。

**CopyTo(T\[\], Int32)**

从**指定数组索引**开始将 ConcurrentBag<T> 元素**复制到现有一维 Array 中**。以下示例代码：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>ConcurrentBag&lt;TempModel&gt; tempModels = new ConcurrentBag&lt;TempModel&gt;();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>tempModels.Add(new TempModel() { Code = "1", Name = "一" });</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>tempModels.Add(new TempModel() { Code = "2", Name = "二" });</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>tempModels.Add(new TempModel() { Code = "3", Name = "三" });</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>TempModel[] temparr = new TempModel[5];</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>tempModels.CopyTo(temparr, 1);</code></td> 
  </tr> 
 </tbody> 
</table>

输出结果为：

![3b6e688f02756d35d595c7d579fde1d6.png][]

**TryPeek(T)**

尝试从 ConcurrentBag<T> 返回一个对象**但不移除**该对象。

**TryTake(T)**

尝试从 ConcurrentBag<T> 中**移除和返回**一个对象。

**ToString()**

返回表示当前对象的字符串。测试值：System.Collections.Concurrent.ConcurrentBag\`1\[Test.ConsoleApp.TempModel\]

**ToArray()**

将 ConcurrentBag<T> 元素复制到新数组。

**GetEnumerator()**

获取当前时间的枚举器。 调用后不影响集合的任何更新。枚举器可以安全地与读取、写入 ConcurrentBag<T> 同时使用。

**GetHashCode()**

获取集合的哈希值。

*参考：[ConcurrentBag][]*

*  [C\# ConcurrentBag的实现原理][C_ ConcurrentBag]*

#### 4、List<T> 和 ConcurrentBag<T> 对比 ####

众所周知，List<T> 集合是非线程安全的，所以我们采用并行编程时会发生丢数据的情况。比如我们通过多线程将一千个对象加入 List<T>，我们最终得到的集合中元素数就会小于一千。

如下测试代码，通过多任务对象 Task 实现将一千个对象加入到 List<T> 中，添加了一千次，但实际上最终的 objects.Count() 值为 913，小于 1000。 但如果将集合名称改成 ConcurrentBag<T>，结果就不会丢失，最终为等于 1000。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>try</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// List&lt;MyObject&gt; objects = new List&lt;MyObject&gt;();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>ConcurrentBag&lt;MyObject&gt; objects = new ConcurrentBag&lt;MyObject&gt;();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Task[] tasks = new Task[1000];</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (int i = 0; i &lt; 1000; i++)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>tasks[i] = Task.Run(() =&gt; </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>objects.Add(new MyObject() { Name = "1", Threadnum = Thread.GetCurrentProcessorId() }));</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Task.WaitAll(tasks); // 等待所有任务完成</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine(objects.Count()); // List&lt;T&gt;：913； ConcurrentBag&lt;T&gt;：1000</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>catch (Exception ex)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>public class MyObject</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>public string Name { get; set; }</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>public int Threadnum { get; set; }</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

[回到顶部][Link 1]

### 二、Parallel 的使用 ###

任务并行库（TPL）支持通过 System.Threading.Tasks.Parallel 类实现数据操作的并行。Parallel.For 或 Parallel.ForEach 编写的循环逻辑与常见的 for 和 foreach 类似，只是增加并行逻辑，来提升效率。TPL 省去了客户端创建线程或列工作项，同时在基本循环中，不需要加锁，TPL 会处理所有低级别的工作。

常用的方法有 Parallel.For、Parallel.ForEach、Parallel.Invoke 等，下面将一一例举。

#### 1、Parallel.For() ####

##### 1.1 重载一：Parallel.For(Int32, Int32, Action<Int32>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>// fromInclusive：开始索引（含） toExclusive：结束索引（不含） body：不允许为 null</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>public static System.Threading.Tasks.ParallelLoopResult For (int fromInclusive, int toExclusive, Action&lt;int&gt; body);</code></td> 
  </tr> 
 </tbody> 
</table>

以下示例使用 For 方法调用 100 个委托，该委托生成随机 Byte 值，并计算其总和：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>ParallelLoopResult result = Parallel.For(0, 100,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>ctr =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>//Random rnd = new Random(ctr * 100000); // public Random(int Seed); // 随机数的种子，若种子相同，多次生成的随机数序列值相同</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Random rnd = new Random();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Byte[] bytes = new Byte[100]; // Byte 数组，每个值的范围为 0~255</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>rnd.NextBytes(bytes); // 生成 100 个 Byte 数值</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int sum = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>foreach (var byt in bytes) // 再将生成的 100 个数值相加</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sum += byt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Iteration {0,2}: {1:N0}", ctr, sum);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Result: Completed Normally");</code></td> 
  </tr> 
 </tbody> 
</table>

##### 1.2 比较执行效率 Parallel.For() 和 for() #####

Paraller.For() 方法类似于 for 循环语句，也是根据入参多次执行同一逻辑操作。使用 Paraller.For() 方法，可以无序的并行运行迭代，而 for 循环只能根据既定的顺序串行运行。

如下实例，对比 Parallel.For() 和 for() 循环的执行效率进行比较：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>// 进行 5 此对比</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (int j = 1; j &lt; 6; j++)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// for()</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"\n第{j}次比较");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>ConcurrentBag&lt;int&gt; bag = new ConcurrentBag&lt;int&gt;();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>var watch = Stopwatch.StartNew();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>watch.Start();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (int i = 0; i &lt; 20000000; i++)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>bag.Add(i);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>watch.Stop();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"串行计算：集合有:{bag.Count},总共耗时：{watch.ElapsedMilliseconds}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Parallel.For()</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>bag = new ConcurrentBag&lt;int&gt;();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>watch = Stopwatch.StartNew();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>watch.Start();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.For(0, 20000000, i =&gt; // i 为整数序列号</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>bag.Add(i);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>watch.Stop();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"并行计算：集合有:{bag.Count},总共耗时：{watch.ElapsedMilliseconds}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

代码总共执行了五次对比，如下图中的耗时比较，很明显，采用并行的 Parallel.For() 远比串行的 for() 效率要高许多。

![cdaa7cba05abf91609f17bbed3f7ab95.png][]

##### 1.3 重载二：Parallel.For(Int32, Int32, Action<Int32,ParallelLoopState>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>// fromInclusive：开始索引（含） toExclusive：结束索引（不含） body：不允许为 null</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>public static ParallelLoopResult For (int fromInclusive, int toExclusive, Action&lt;int, ParallelLoopState&gt; body);</code></td> 
  </tr> 
 </tbody> 
</table>

此重载增加了 System.Threading.Tasks.ParallelLoopState 循环状态参数，从而使得我们可以通过循环状态来控制并行循环的运行。

以下实例，执行 100 次迭代，在随机数 breakIndex 指示的一次迭代时进行中断操作，调用完 Break() 方法后，循环状态的 ShouldExitCurrentIteration 属性值就是 true，然后进入判断`if (state.LowestBreakIteration < i)`，当**当前迭代序号**大于中断时的序号，就直接返回，不再进行后续操作。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>var rnd = new Random();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int breakIndex = rnd.Next(1, 11);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Will call Break at iteration {breakIndex}\n");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>var result = Parallel.For(1, 101, (i, state) =&gt; // 实际执行的是 1 ~ 100，不包含 101</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Beginning iteration {i} {Thread.GetCurrentProcessorId()}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int delay;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>lock (rnd)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>delay = rnd.Next(1, 1001);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Thread.Sleep(delay);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (state.ShouldExitCurrentIteration)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (state.LowestBreakIteration &lt; i)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>return;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (i == breakIndex) // 8</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Break in iteration {i}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>state.Break();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Completed iteration {i} {Thread.GetCurrentProcessorId()}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (result.LowestBreakIteration.HasValue)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"\nLowest Break Iteration: {result.LowestBreakIteration}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>else</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"\nNo lowest break iteration.");</code></td> 
  </tr> 
 </tbody> 
</table>

如下是当索引值为 9 时的处理过程：（当迭代序号为 9 时，执行 Break()，此之前已经开始迭代执行的大于 9 的迭代，均直接退出，只有开始没有结束）

![a32db86150d3c4c2db2b285d625cebbf.png][]

##### 1.4 重载三：Parallel.For(Int32, Int32, ParallelOptions, Action<Int32,ParallelLoopState>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>// fromInclusive：开始索引（含） toExclusive：结束索引（不含） body：不允许为 null</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>public static ParallelLoopResult For (int fromInclusive, int toExclusive, ParallelOptions parallelOptions, Action&lt;int, ParallelLoopState&gt; body);</code></td> 
  </tr> 
 </tbody> 
</table>

此重载在执行 for 循环时，可以配置循环选项 ParallelOptions。

下边是一个实例，通过配置 ParallelOptions 的 CancellationToken 属性，使得循环支持手动取消：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>CancellationTokenSource cancellationSource = new CancellationTokenSource();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>ParallelOptions options = new ParallelOptions();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>options.CancellationToken = cancellationSource.Token;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>try</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>ParallelLoopResult loopResult = Parallel.For( 0, 10, options,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>(i, loopState) =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Start Thread={0}, i={1}", Thread.CurrentThread.ManagedThreadId, i);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (i == 5) // 模拟某次迭代执行时，取消循环</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>cancellationSource.Cancel();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (int j = 0; j &lt; 10; j++)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Thread.Sleep(1 * 200); // 模拟耗时任务</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (loopState.ShouldExitCurrentIteration) // 判断循环是否已经取消执行</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>return;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Finish Thread={Thread.CurrentThread.ManagedThreadId}, i={i}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (loopResult.IsCompleted)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("All iterations completed successfully. THIS WAS NOT EXPECTED.");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>catch (AggregateException aex) // 注意：AggregateException 为并行中专用的异常信息集合</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Parallel.For has thrown an AggregateException. THIS WAS NOT EXPECTED.\n{aex}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>//foreach (var item in aex.InnerExceptions) // 可以通过循环将全部信息记录下来</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>//{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Console.WriteLine(item.InnerException.Message + " " + item.GetType().Name);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>//}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>//aex.Handle(p =&gt; // 如果想往上级抛，需要使用 Handle 方法处理一下</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>//{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// if (p.InnerException.Message == "my god！Exception from childTask1 happend！")</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// return true;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// else</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// return false; // 返回 false 表示往上继续抛出异常</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>//});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>catch (OperationCanceledException ocex) // 专门用于取消循环异常的捕捉</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"An iteration has triggered a cancellation. THIS WAS EXPECTED.\n{ocex}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>finally</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>cancellationSource.Dispose();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

如下图中的输出，所有迭代任务都未完成，主要是因为耗时操作执行完成之前，循环就取消了，在`if (loopState.ShouldExitCurrentIteration)`判断时，均为 true 就直接返回了。

![71167e763315f48d3d0a10ec09de2506.png][]

##### 1.5 重载四：For<TLocal>(Int32, Int32, ParallelOptions, Func<TLocal>, Func<Int32,ParallelLoopState,TLocal,TLocal>, Action<TLocal>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>public static ParallelLoopResult For&lt;TLocal&gt; (int fromInclusive, int toExclusive, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>ParallelOptions parallelOptions, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Func&lt;TLocal&gt; localInit, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Func&lt;int,ParallelLoopState,TLocal,TLocal&gt; body, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Action&lt;TLocal&gt; localFinally);</code></td> 
  </tr> 
 </tbody> 
</table>

以下示例使用线程局部变量来计算许多冗长操作的结果之和。 此示例将并行度限制为 4。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int result = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int N = 1000000;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.For(</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>0, N,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 限制最多 4 个并行任务</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>new ParallelOptions { MaxDegreeOfParallelism = 4 },</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Func&lt;TLocal&gt; 初始化本地变量，本地变量是线程独立变量</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>() =&gt; 0,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Func&lt;Int32,ParallelLoopState,TLocal,TLocal&gt; 迭代操作</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>(i, loop, localState) =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (int ii = 0; ii &lt; 10000; ii++) ;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>return localState + 1;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>},</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>localState =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Interlocked.Add(ref result, localState)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("实际运算结果： {0}. 目标值： 1000000", result);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

如下图输出结果：

![8546bdc2d7685cda23cd976ba5989f2e.png][]

*参考：[Parallel.For 方法 (System.Threading.Tasks) | Microsoft Learn][Parallel.For _ _System.Threading.Tasks_ _ Microsoft Learn]*

*关于 ParallelOptions 详见：[ParallelOptions 类 (System.Threading.Tasks) | Microsoft Learn][ParallelOptions _ _System.Threading.Tasks_ _ Microsoft Learn]*

#### 2、Parallel.ForEach() ####

##### 2.1 重载一：Parallel.ForEach<TSource>(IEnumerable<TSource>, Action<TSource>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>public static ParallelLoopResult ForEach&lt;TSource&gt; (IEnumerable&lt;TSource&gt; source, Action&lt;TSource&gt; body);</code></td> 
  </tr> 
 </tbody> 
</table>

执行 ForEach 操作，在处理关于 IEnumerable 集合的任务时，可并行运行迭代。

如下代码块，简单的将一个整数数组，输出到控制台：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int[] ints = { 11, 12, 13, 14, 15, 16, 17, 18, 19 };</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>ParallelLoopResult result = Parallel.ForEach(ints,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>i =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine(i);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

从输出结果看，ForEach 操作是无序的：

![b14dd79ccc7e81f5dc8ff2647f0e093f.png][]

##### 2.2 重载二：ForEach<TSource>(IEnumerable<TSource>, ParallelOptions, Action<TSource,ParallelLoopState,Int64>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>public static ParallelLoopResult ForEach&lt;TSource&gt; (IEnumerable&lt;TSource&gt; source, ParallelOptions parallelOptions, Action&lt;TSource,ParallelLoopState,long&gt; body);</code></td> 
  </tr> 
 </tbody> 
</table>

执行具有 64 位索引（标识待循环集合的顺序）的 foreach 操作，其中在 IEnumerable 上可能会并行运行迭代，而且可以配置循环选项，可以监视和操作循环的状态。

如下示例代码，设置并行任务数为 5，在索引为 6 的任务执行过程中中断循环，看下输出结果：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 创建一个集合，其中包含一些数字</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>var numbers = new int[] { 10, 11, 12, 13, 14, 15, 16, 17, 18, 19 };</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 使用 ParallelOptions 选项设置并行处理的行为</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>var parallelOptions = new ParallelOptions</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>MaxDegreeOfParallelism = 5</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>};</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.ForEach(numbers, parallelOptions, (source, loopState, index) =&gt; // index：集合中对象的从 0 开始的序号</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 在此处编写并行处理逻辑</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"开始--Index: {index}, Value: {source}, ThreadId: {Thread.GetCurrentProcessorId()}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (loopState.ShouldExitCurrentIteration)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>return;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Thread.Sleep(200);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>if (index == 6)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>loopState.Break();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"结束++Index: {index}, Value: {source}, ThreadId: {Thread.GetCurrentProcessorId()}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

如下图输出结果，一次性开始 5 个并行任务，当第 6 个任务进入时，中断循环。

由于操作是无序的，所以在中断之前可能索引在 6 之后的已经开始或者已经执行完成，如下图 8、9 已经执行完毕，7尚未执行。

注意，若允许并行的任务数少时，可能 6 之后的任务都还没来得及开始，另外，每次执行的结果不同。

![92c98fc684d86957998b4ed7c8353490.png][]

##### 2.3 重载三：Parallel.ForEach<TSource>(Partitioner<TSource>, Action<TSource>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>public static ParallelLoopResult ForEach&lt;TSource&gt; (System.Collections.Concurrent.Partitioner&lt;TSource&gt; source, Action&lt;TSource&gt; body);</code></td> 
  </tr> 
 </tbody> 
</table>

此重载的独到之处，就是可以**将数据进行分区，每一个小区内实现串行计算**，分区采用 Partitioner.Create() 实现。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>long sum = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>long sumtop = 10000000;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Stopwatch sw = Stopwatch.StartNew();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.ForEach(Partitioner.Create(0, sumtop), (range) =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>long local = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (long i = range.Item1; i &lt; range.Item2; i++)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>local += i;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Interlocked.Add(ref sum, local); // Interlocked：为由多个线程共享的变量提供原子操作 Add()：求和后替换原来的数值，相当于 +=</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw.Stop();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Partitioner.Create() 分区方式执行效率： result = {sum}, time = {sw.ElapsedMilliseconds} ms");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 输出：</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Partitioner.Create() 分区方式执行效率： result = 49999995000000, time = 8 ms</code></td> 
  </tr> 
 </tbody> 
</table>

关于分区的创建方法 Partitioner.Create(0, Int64)

*  指定了分区的范围，就是 0 ~ Int64；
 *  参数中并没有指定分多少个区，默认是系统自动判断执行的。
 *  还可以指定分区，做法就是`Partitioner.Create(0, 3000000, Environment.ProcessorCount)`，其中 Environment.ProcessorCount 参数，就对应当前计算机**逻辑处理器**的数量。

##### 2.4 重载四：ForEach<TSource,TLocal>(IEnumerable<TSource>, Func<TLocal>, Func<TSource,ParallelLoopState,TLocal,TLocal>, Action<TLocal>) #####

执行具有线程本地数据的 foreach 操作，其中在 IEnumerable 上可能会并行运行迭代，而且可以监视和操作循环的状态。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>public static ParallelLoopResult ForEach&lt;TSource,TLocal&gt; (IEnumerable&lt;TSource&gt; source, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Func&lt;TLocal&gt; localInit,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Func&lt;TSource,ParallelLoopState,TLocal,TLocal&gt; body, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Action&lt;TLocal&gt; localFinally);</code></td> 
  </tr> 
 </tbody> 
</table>

如下示例，将全部整数逐个输出并且最后在输出他们之和：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 全部值的和为 40</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int[] input = { 4, 1, 6, 2, 9, 5, 10, 3 };</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>int sum = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>try</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.ForEach(</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// IEnumerable&lt;TSource&gt; 可枚举的数据源</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>input,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Func&lt;TLocal&gt; 用于返回每个任务的【本地数据的初始状态】的函数委托</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 本示例中的目的就是将 TLocal localSum 的值在每次迭代都赋值为 0</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>() =&gt; 0,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Func&lt;TSource,ParallelLoopState,TLocal,TLocal&gt; 将为每个迭代调用一次的委托</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>(n, loopState, localSum) =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>localSum += n;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"Thread={Thread.CurrentThread.ManagedThreadId}, n={n}, localSum={localSum}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>return localSum;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>},</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Action&lt;TLocal&gt; 用于对每个任务的本地状态执行一个最终操作的委托</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 此示例中的作用是将每个值逐一求和，并返回 sum</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>(localSum) =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Interlocked.Add(ref sum, localSum)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("\nSum={0}", sum);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>catch (AggregateException e)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Parallel.ForEach has thrown an exception. This was not expected.\n{0}", e);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

如下输出结果，其中 localSum 在每个线程中初始值都是 0，在其他线程中参与的求和运算，不影响当前线程。

![6727921844303125e0794de4d150cfc5.png][]

##### 2.5 比较执行效率 for、Parallel.For()、Parallel.For()+TLocal、Parallel.ForEach(Partitioner.Create(), Action<TSource>) #####

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Stopwatch sw = null;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>long sum = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>long sumtop = 10000000;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 常规 for 循环</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw = Stopwatch.StartNew();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (long i = 0; i &lt; sumtop; i++)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sum += i;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw.Stop();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"result = {sum}, time = {sw.ElapsedMilliseconds} ms --常规 for 循环");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Parallel.For() 方式</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sum = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw = Stopwatch.StartNew();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.For(0L, sumtop,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>(item) =&gt; Interlocked.Add(ref sum, item));</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw.Stop();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"result = {sum}, time = {sw.ElapsedMilliseconds} ms --Parallel.For() 方式");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Parallel.For() + TLocal</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sum = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw = Stopwatch.StartNew();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.For(</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>0L, sumtop, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>() =&gt; 0L, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>(item, state, prevLocal) =&gt; </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>prevLocal + item, </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>local =&gt; </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Interlocked.Add(ref sum, local));</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw.Stop();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"result = {sum}, time = {sw.ElapsedMilliseconds} ms --Parallel.For() + locals 方式");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// Partitioner.Create() 分区方式</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sum = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw = Stopwatch.StartNew();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.ForEach(Partitioner.Create(0L, sumtop), (range) =&gt;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>long local = 0;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>for (long i = range.Item1; i &lt; range.Item2; i++)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>local += i;</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Interlocked.Add(ref sum, local);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>sw.Stop();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"result = {sum}, time = {sw.ElapsedMilliseconds} ms --Partitioner.Create() 分区方式");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

如下输出结果，效率最高的显然是自动分区的方式，比常规的 for 循环块将近一倍。最慢的是 Parallel.For() 方式，由于加锁求和导致上下文频繁切换比较耗时，因此这种求和的计算模式不适用。

![5bc6d5e634cd953d54d09e9229d1cc50.png][]

*参考：[Parallel.ForEach 方法 (System.Threading.Tasks) | Microsoft Learn][Parallel.ForEach _ _System.Threading.Tasks_ _ Microsoft Learn]*

#### 3、Parallel.ForEachAsync() ####

Parallel.ForEachAsync() 是在 .NET 6 中新增的一个 API，是 Parallel.ForEach() 的异步版本。[Parallel.ForEachAsync 方法 (System.Threading.Tasks) | Microsoft Learn][Parallel.ForEachAsync _ _System.Threading.Tasks_ _ Microsoft Learn]

下面简单说明一下 Parallel.ForEach() 和 Parallel.ForEachAsync() 的区别。

*  Parallel.ForEach() 是在默认多个或指定的个数的线程下执行的。而 Parallel.ForEachAsync() 不一定是多线程的，强调的是异步而已。
 *  若目标集合必须按照顺序执行，则不能选用 Parallel.ForEach() 方法，因为它是无序执行的。
 *  当待处理的数据量很大或者执行过程比较耗时，则选用多线程执行的 Parallel.ForEach() 方法更好。

下面是一个关于重载 ForEachAsync<TSource>(IAsyncEnumerable<TSource>, ParallelOptions, Func<TSource,CancellationToken,ValueTask>) 的一个简单示例代码：

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static async Task Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>var nums = Enumerable.Range(0, 10).ToArray();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>await Parallel.ForEachAsync(</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>nums,</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>new ParallelOptions { MaxDegreeOfParallelism = 3 }, // 配置最多同时分配三个线程</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>async (i, token) =&gt; // Func&lt;TSource,CancellationToken,ValueTask&gt; // 其中 ValueTask 提供异步操作的可等待结果，指的是下文 await 的内容</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"开始迭代任务 {i} ThreadId:{Thread.GetCurrentProcessorId()}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// public static Task Delay(int millisecondsDelay, CancellationToken cancellationToken)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>// 在指定毫秒后，调用 token 取消当前任务</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>await Task.Delay(1000, token); </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine($"完成迭代任务 {i}");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>});</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Finished!");</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

*详情可参考：[Parallel.ForEachAsync 方法 (System.Threading.Tasks) | Microsoft Learn][Parallel.ForEachAsync _ _System.Threading.Tasks_ _ Microsoft Learn] ；* [Parallel.ForEachAsync Deep Dive :: Greg Bair][Parallel.ForEachAsync Deep Dive _ Greg Bair]

#### 4、Parallel.Invoke() ####

尽可能并行执行提供的每个操作。

##### 4.1 两个重载：Invoke(Action\[\])、Invoke(ParallelOptions, Action\[\]) #####

下面是一个运用 Invoke(Action\[\]) 重载的示例，分别加入了三个操作，然后看执行结果。第二个重载是在第一个重载的基础上加了并行选项 ParallelOptions 就不在赘述了。

<table> 
 <tbody> 
  <tr> 
   <td></td> 
   <td><code>static void Main(string[] args)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>try</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Parallel.Invoke(</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>BasicAction, // 第一个操作 - 静态方法</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>() =&gt; // 第二个操作 - 箭头函数</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Method=beta, Thread={0}", Thread.CurrentThread.ManagedThreadId);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>},</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>delegate () // 第三个操作 - 委托函数</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Method=gamma, Thread={0}", Thread.CurrentThread.ManagedThreadId);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>catch (AggregateException e)</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("An action has thrown an exception. THIS WAS UNEXPECTED.\n{0}", e.InnerException.ToString());</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.ReadLine();</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>static void BasicAction()</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>{ </code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>Console.WriteLine("Method=alpha, Thread={0}", Thread.CurrentThread.ManagedThreadId);</code></td> 
  </tr> 
  <tr> 
   <td></td> 
   <td><code>}</code></td> 
  </tr> 
 </tbody> 
</table>

由输出结果可知，三个操作是无序的、多线程执行的。

![e59082c808dc5a136670ee037a65c2ec.png][]

*两个参考：[Parallel.Invoke 方法 (System.Threading.Tasks) | Microsoft Learn][Parallel.Invoke _ _System.Threading.Tasks_ _ Microsoft Learn] [Parallel的使用][Parallel]*

[回到顶部][Link 1]

### 三、简单总结一下下 ###

实际上看的资料再多，如果没用到实际开发当中就是无用功，下边简单总结一下吧。

由本文 [1.2 比较执行效率 Parallel.For() 和 for()][1.2 _ Parallel.For_ _ for] 中可知：

*  对于**大批量耗时且顺序要求不高**的场景可以采用 Parallel.For() 方法，如果对次序有依赖，则只能采用常用的 for 循环。
 *  对于**操作简单**的循环操作，Parallel.For() 就不太适合了，因为多线程操作涉及到上下文的切换，过多的切换场景会严重影响程序运行的效率。

由本文 [2.5 比较执行效率 for、Parallel.For()、Parallel.For()+TLocal、Parallel.ForEach(Partitioner.Create(), Action<TSource>)][2.5 _ for_Parallel.For_Parallel.For_TLocal_Parallel.ForEach_Partitioner.Create_ Action_TSource] 中可知：

*  由于示例中的操作比较简单，此时 Parallel.For() 上下文的的切换耗时以及加锁的缺点就凸现了，效率最差。
 *  使用线程本地变量（TLocal）的 Parallel.For() 可以避免将大量的访问同步为共享状态的开销，所以可以看到效率就高很多。*可参考：[编写具有线程局部变量的 Parallel.For 循环][Parallel.For]*
 *  分区循环操作 Partitioner.Create(0, Int64) 方法的效率最高，因为事先给待处理的任务进行了分区，分区内串行，避免了过多的上下文切换耗时。

[Link 1]: https://www.cnblogs.com/czzj/p/ParallelAndConcurrentBag.html#_labelTop
[3b6e688f02756d35d595c7d579fde1d6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/0af872a0ab3d438caf601b49f18d1c67.png
[ConcurrentBag]: https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.concurrentbag-1?view=net-5.0
[C_ ConcurrentBag]: https://www.cnblogs.com/InCerry/p/9497729.html
[cdaa7cba05abf91609f17bbed3f7ab95.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/dbe44e589cf94d2bb939348449b88d06.png
[a32db86150d3c4c2db2b285d625cebbf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/beab133a843242ecaaa2edf9fcf4c79d.png
[71167e763315f48d3d0a10ec09de2506.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/8b62a3f2f07243a195c2b4534526d7b5.png
[8546bdc2d7685cda23cd976ba5989f2e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/d4dcb6b02ead45c5bb7e4325e6a078d4.png
[Parallel.For _ _System.Threading.Tasks_ _ Microsoft Learn]: https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.for?view=net-7.0
[ParallelOptions _ _System.Threading.Tasks_ _ Microsoft Learn]: https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.paralleloptions?view=net-7.0
[b14dd79ccc7e81f5dc8ff2647f0e093f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/cb84cd2dd13c487aa9dacf0d72c8e407.png
[92c98fc684d86957998b4ed7c8353490.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/35d22ba90d594a4389f4f14baa650d42.png
[6727921844303125e0794de4d150cfc5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/7783f38c7b3a40c08afd07ac525db4ad.png
[5bc6d5e634cd953d54d09e9229d1cc50.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/92316e2e0ba94ec1917890736362a7e9.png
[Parallel.ForEach _ _System.Threading.Tasks_ _ Microsoft Learn]: https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.foreach?view=net-7.0
[Parallel.ForEachAsync _ _System.Threading.Tasks_ _ Microsoft Learn]: https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-7.0
[Parallel.ForEachAsync Deep Dive _ Greg Bair]: https://www.gregbair.dev/posts/parallel-foreachasync/
[e59082c808dc5a136670ee037a65c2ec.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/aa30b2e7812447b9a6b5204b42fc790f.png
[Parallel.Invoke _ _System.Threading.Tasks_ _ Microsoft Learn]: https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel.invoke?view=net-7.0
[Parallel]: https://www.cnblogs.com/huangxincheng/archive/2012/04/02/2429543.html
[1.2 _ Parallel.For_ _ for]: https://www.cnblogs.com/czzj/p/ParallelAndConcurrentBag.html#_label2_0_1
[2.5 _ for_Parallel.For_Parallel.For_TLocal_Parallel.ForEach_Partitioner.Create_ Action_TSource]: https://www.cnblogs.com/czzj/p/ParallelAndConcurrentBag.html#_label2_1_4
[Parallel.For]: https://learn.microsoft.com/zh-cn/dotnet/standard/parallel-programming/how-to-write-a-parallel-for-loop-with-thread-local-variables