数据并行处理

1. 介绍

今天咱们来看看如何尽可能快地处理大量数据，利用你电脑（或服务器）上的所有可用 CPU 核。为此我们会用到 System.Threading.Tasks.Parallel 里的类，特别是 Parallel.For 和 Parallel.ForEach。

如果任务是纯 CPU-bound 怎么办？

传统的 for 或 foreach 循环会一个接一个地处理元素。简单又可靠。但在多核处理器上，循环只会用到一个核，其他核白等。为什么不把数组的不同部分分配到不同核，让它们同时处理呢？

示例：

// 计算从 1 到 N 的平方和
long sum = 0;
for (int i = 1; i <= 1_000_000; i++)
{
    sum += i * i;
}

这段代码很简单，但顺序执行。如果把任务分散到各个核会怎样？

认识家族： Parallel.For 和 Parallel.ForEach

这是什么？

• Parallel.For — 像普通的 for，只是把工作拆成若干部分并自动分配到线程，利用所有可用核。
• Parallel.ForEach — 像普通的 foreach 遍历集合，但并行执行每个元素的处理。

官方文档：

• Parallel.For
• Parallel.ForEach

为什么方便？

你不需要自己创建、启动和管理线程。框架在底层处理这些细节。你写的代码看起来像普通循环，但并行会在背后自动发生。

2. 语法：基本示例

Parallel.For

long total = 0;
Parallel.For(1, 1_000_001, i =>
{
    // 这个 lambda 可能会被不同线程同时执行
    Interlocked.Add(ref total, i * i); // 为了避免竞态
});
Console.WriteLine($"平方和: {total}");

注意：变量 total 我们通过 Interlocked.Add 更新 —— 以避免数据竞争。

Parallel.ForEach

var numbers = Enumerable.Range(1, 10_000_000).ToArray();
long sum = 0;

Parallel.ForEach(numbers, num =>
{
    Interlocked.Add(ref sum, num * num); // 安全地累加
});
Console.WriteLine($"平方和: {sum}");

从内部看（示意图）

+-------------------+
|集合/范围          |
+---------+---------+
          |
          v
  +----------------------+
  |   Parallel.ForEach   |
  +----------+-----------+
             |
        +----+----+----+----+
        |         |         |
        v         v         v
  Task #1    Task #2    Task #3   ... (可用的核)
        |         |         |
     +--+----+  +--+-----+  +--+-----+
     |处理    |  |处理     |  |处理     |
     +-------+  +--------+  +--------+
        \         |         /
         +--------+--------+
                  |
                  v
               结果

3. 分析大文件（CPU-bound 处理）

假设我们有一个包含数万行的文本文件 —— 比如每行是一个数字。需要读取文件，把每个数平方后求平方和。

同步版本

string[] lines = File.ReadAllLines("numbers.txt");
long sum = 0;

foreach (var line in lines)
{
    if (long.TryParse(line, out long n))
    {
        sum += n * n;
    }
}
Console.WriteLine($"平方和: {sum}");

使用 Parallel.For 的并行版本

string[] lines = File.ReadAllLines("numbers.txt");
long sum = 0;

Parallel.For(0, lines.Length, i =>
{
    if (long.TryParse(lines[i], out long n))
    {
        Interlocked.Add(ref sum, n * n);
    }
});
Console.WriteLine($"平方和: {sum}");

变化：我们把普通循环换成并行循环，并通过 Interlocked.Add 增加 sum —— 以避免线程间冲突。

4. 底层发生了什么？

当你调用 Parallel.For 或 Parallel.ForEach 时，.NET 会自动把工作切分成多个片段，并用线程池把它们分配到可用的处理器核上。每个片段在自己的线程里独立处理。

优点：如果你有 4 个核，工作可能接近 4 倍加速（前提是任务不依赖外部资源，并且不受其它瓶颈限制，比如内存或磁盘读取速度）。

比较执行时间

var numbers = Enumerable.Range(1, 100_000_000).ToArray();
long sumSync = 0;
var sw = System.Diagnostics.Stopwatch.StartNew();

foreach (var n in numbers)
    sumSync += n * n;

sw.Stop();
Console.WriteLine($"Sync: {sw.ElapsedMilliseconds} ms, 总和: {sumSync}");

long sumParallel = 0;
sw.Restart();

Parallel.ForEach(numbers, n =>
    Interlocked.Add(ref sumParallel, n * n)
);

sw.Stop();
Console.WriteLine($"Parallel: {sw.ElapsedMilliseconds} ms, 总和: {sumParallel}");

自己试试！ 在性能强劲的机器上可能能看到数倍加速，但具体效果取决于任务和瓶颈。

5. 有用的细节

控制并行度

有时候你需要限制使用的线程数量（比如不想把系统压垮）。这时可以用 MaxDegreeOfParallelism：

using System.Threading.Tasks;
long sum = 0;
var options = new ParallelOptions { 
    MaxDegreeOfParallelism = 2 
};
Parallel.For(0, 100, options, i =>
{
    Interlocked.Add(ref sum, i * i);
});
Console.WriteLine($"平方和: {sum}");

什么时候有用：如果你知道部分计算主要受磁盘限制而不是 CPU，那么把线程数调小，观察对性能的影响。

什么时候用并行循环

拓展：Parallel 还能做什么？

Parallel.Invoke() — 同时运行几个独立的方法：

static void DoTask1() => Console.WriteLine("任务 1 已完成");
static void DoTask2() => Console.WriteLine("任务 2 已完成");
static void DoTask3() => Console.WriteLine("任务 3 已完成");

Parallel.Invoke(
    () => DoTask1(),
    () => DoTask2(),
    () => DoTask3()
);

每个方法会尽可能在自己的核上执行。

实际应用

• 图像处理：并行处理不同块（例如应用滤镜）。
• 数组计算：金融计算、建模（按场景评估投资组合）。
• 处理大日志：在多个核上并行查找和聚合。
• 机器学习：把任务拆成独立子任务（数据 batch、特征工程）。

当然，在面试时你不仅能说出什么是并行循环，还能坦诚地说明它们的优缺点。

6. 使用 Parallel.For 和 Parallel.ForEach 时常见错误

错误 #1：忽视数据竞争。
在没有使用 Interlocked 或 lock 的情况下更新共享变量，会因为线程同时访问导致结果不正确。

错误 #2：用于 I/O-bound 任务。
并行循环不会加速依赖磁盘或网络的任务，反而可能因为额外开销变慢。

错误 #3：假设有执行顺序。
并行循环不保证元素处理顺序，如果逻辑依赖顺序可能会出错。

错误 #4：忽视副作用。
在并行循环中修改共享状态（例如集合）可能导致错误，除非使用线程安全的数据结构。