【C#】并行编程实战：使用 PLINQ（3）

news2024/7/6 18:35:44

PLINQ 是语言集成查询（Language Integrate Query , LINQ）的并行实现（P 表示并行）。本章将继续介绍其编程的各个方面以及与之相关的一些优缺点。

本文的主要内容为 PLINQ 中的组合并行和顺序 LINQ 查询、取消 PLINQ 查询、使用 PLINQ 进行并行编程时要考虑的事项和影响 PLINQ 性能的因素。

本教程对应学习工程：魔术师Dix / HandsOnParallelProgramming · GitCode

6、组合并行和顺序 LINQ 查询

有时，我们可能会希望顺序执行运算符，这时就可以使用 AsSequential 方法强制 PLINQ 按顺序执行。一旦该方法用于任何并行查询，之后的运算符就会按照顺序执行。

这个示例已经在 2.2、顺序查询里有所展示，这里就不再赘述。

如何：合并并行和顺序 LINQ 查询 | Microsoft Learn详细了解：如何：合并并行和顺序 LINQ 查询https://learn.microsoft.com/zh-cn/dotnet/standard/parallel-programming/how-to-combine-parallel-and-sequential-linq-queries

7、取消 PLINQ 查询

可以使用 CancellationTokenSource 和 CancellationToken 类取消 PLINQ 查询。

CancellationToken （取消令牌）将使用 WithCancellation 子语句传递到 PLINQ 查询，然后可以调用 CancellationToken.Cancel 方法取消查询操作。在取消之后，将抛出 OperationCanceledException 异常。

代码示例如下：

        private void RunWithCancellationTokenSource()
        {
            //由外部设置最大并行度。
            int degreeOfParallelism = commonPanel.GetInt32Parameter();
            Debug.Log($"设置并行度为：{degreeOfParallelism}");

            //使用取消令牌
            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();

            var range = Enumerable.Range(1, 10000).AsParallel()
                .WithCancellation(cancellationTokenSource.Token)
                .WithDegreeOfParallelism(degreeOfParallelism)
                .Select(x =>x);

            try
            {
                range.ForAll(x =>
                {
                    if (x == 5)
                    {
                        cancellationTokenSource.Cancel();
                        Debug.Log("Cancel PLINQ !");
                    }
                    else
                    {
                        Debug.Log($"PLINQ Is Running : {x}");
                    }
                });
            }
            catch (AggregateException ex)
            {
                foreach (var item in ex.InnerExceptions)
                {
                    Debug.LogError(item.InnerException);
                }
            }
        }

在我的电脑上运行结果如下：

仍然执行了 64 次 Select 语句，并且最后抛出了异常：

值得一提的是，如果将整段代码放到 Task.Run 语句里，则是不会抛出异常的，因为 TryCatch 是不会跨线程生效的。

8、使用 PLINQ 进行并行编程时要考虑的事项

在大多数情况下，PLINQ 的性能比费并行同类产品 LINQ 要快得多。但是，它也存在一些性能开销，这与在并行化 LINQ 时进行的分区和合并有关。以下是使用 PLINQ 时需要考虑的一些事项：

合并执行并不意味着一定更快：并行化本身也需要开销，因此，除非你的源集合很大，或者操作需要大量的计算，否则按顺序执行这些操作更有意义。可以通过衡量顺序查询和并行查询的性能来做出明智的决定。
避免涉及原子性的 I/O 操作：在 PLINQ 内部应避免所有涉及写入文件系统、数据库、网络或共享内存位置的 I/O 操作。其原因在于，这些方法不是线程安全（Thread-Safe）的，因此使用他们可能会导致异常。一种解决方案是使用同步原语，但这也会大大降低性能。
查询并不一定总是并行运行的：PLINQ 中的并行化是由公共语言运行时（CLR）做出的决定。即便在查询中调用了 AsParallel 方法，它也不保证采用并行路径，同样有可能顺序运行。

9、影响 PLINQ 性能的因素

PLINQ 的主要目的是通过拆分任务并按并行方式执行来加速查询执行的。但是，有很多因素都会影响 PLINQ 的性能。其中包括与分块有关的同步开销，以及调度和收集线程结果的开销。

在理想的并行场景中，线程不必共享状态，也不必担心执行顺序。

9.1、并行度

由于 TPL 确保多个任务可以在多个内核上同时执行，因此我们可以使用更多数量的内核，从而显著提高性能。性能的提高可能不会是指数级的，并且在调整性能时，我们也应该尝试在具有多个内核的不同系统上运行比较结果。

9.2、合并选项

在某些应用场景中，结果经常变化，并且用户希望尽快看到结果而无需等待。在这些情况下，合并选项可以显著改善用户体验。PLINQ 默认选项是缓冲结果（AutoBuffered），然后将其合并以返回用户。我们可以通过选择适当的合并选项来修改此行为。

9.3、分区类型

我们应始终检查分区的工作项目是否平衡。对于不平和的工作项目，可以引入自定义分区以提高性能。

PLINQ 和 TPL 的自定义分区程序 | Microsoft Learn详细了解：PLINQ 和 TPL 的自定义分区程序https://learn.microsoft.com/zh-cn/dotnet/standard/parallel-programming/custom-partitioners-for-plinq-and-tpl PS：我个人认为自定义分区程序是一个很重要的功能点，但是上一章并没有详细深入地讲解，我也没有仔细学习这部分。我不知道后面的章节是否会详细讲解自定义分区，如果没有的话，我会考虑单独开一章来学习一下如何编写自定义分区程序。

如何：实现动态分区 | Microsoft Learn详细了解：如何实现动态分区https://learn.microsoft.com/zh-cn/dotnet/standard/parallel-programming/how-to-implement-dynamic-partitions

9.4、确定是保持顺序还是转向并行

我们应该始终计算出每个工作项以及整个操作的整体计算成本，以便可以决定是保持顺序执行还是转向并行。由于分区、调度等产生的额外开销，并行查询不一定是最快的。

计算成本的公式：

计算成本 = 执行1个工作项目的而成本 * 工作项目的总数 + 并行开销

PLINQ 决定是采用顺序执行还是并行执行取决于查询中运算符的组合。

可以参考使用并发可视化工具来辅助性能衡量：

并发可视化工具 - Visual Studio (Windows) | Microsoft Learn使用并发可视化工具查看多线程应用中显示线程计时的图形，从而帮助你解决性能问题。https://learn.microsoft.com/zh-cn/visualstudio/profiling/concurrency-visualizer?view=vs-2022

9.5、操作顺序

PLINQ 可为无序集合提供更好的性能，因为使集合按照有序方式执行是会产生性能成本的。该性能成本包括分区、调度和收集结果，以及调用 GroupJoin 和过滤器。

简单地说，就是能无序就无序，尽量不要使用类似 AsOrdered 的顺序执行方法。

9.6、使用 ForAll

调用 ToList 、 ToArray 或在循环中枚举结果时，实际上是强制 PLINQ 将来自所有并行线程的结果合并为单个数据结构，这也会产生性能开销。因此，如果只想对一组项目执行某些操作，最好使用 ForAll 方法。

9.7、强制并行

PLINQ 不保证每次都以并行方式执行，我们可以使用 WithExecutionMode 来对此进行控制。示例代码如下：

        private void ForceToParallel()
        {
            var range = Enumerable.Range(1, 10);
            var squares = range.AsParallel()
                .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                .Select(x => x * x)
                .ToList();

            squares.ForEach(x =>
            {
                Debug.Log(x);
            });
        }

PS ：WithExecutionMode 会被 AsOrderd 之类的函数给覆盖掉。