C Sharp 数据并行

C#数据并行

介绍

数据并行是并行编程的一种范式，指对数据集中的每个元素独立执行相同的操作。在C#中，数据并行主要通过System.Threading.Tasks.Parallel类和PLINQ（Parallel LINQ）实现。它适用于计算密集型任务，能够显著提高多核CPU的利用率。

数据并行的核心特点是：

• 将数据集分割为多个独立分区
• 每个分区由不同的线程处理
• 最终合并处理结果

与任务并行（处理不同任务）不同，数据并行专注于对同一操作应用到大容量数据上。

Parallel.For 和 Parallel.ForEach

C#提供了两种主要的数据并行结构：

Parallel.For

用于并行执行for循环迭代，语法类似常规for循环：

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        int[] data = { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
        
        Parallel.For(0, data.Length, i => 
        {
            data[i] = data[i] * 2;
            Console.WriteLine($"Processing index {i} on thread {Task.CurrentId}");
        });
        
        Console.WriteLine("Result:");
        foreach (var item in data)
        {
            Console.Write(item + " ");
        }
    }
}

输出示例（顺序可能不同）：

Processing index 2 on thread 1
Processing index 0 on thread 3
Processing index 5 on thread 2
Processing index 8 on thread 4
Processing index 1 on thread 5
Processing index 6 on thread 6
Processing index 3 on thread 7
Processing index 7 on thread 8
Processing index 4 on thread 9
Processing index 9 on thread 10
Result:
2 4 6 8 10 12 14 16 18 20 

Parallel.ForEach

用于并行处理集合中的每个元素：

List<string> urls = new List<string> 
{
    "https://example.com/page1",
    "https://example.com/page2",
    // ...更多URL
};

Parallel.ForEach(urls, url => 
{
    DownloadData(url); // 假设的下载方法
});

并行选项

可以通过ParallelOptions配置并行行为：

var options = new ParallelOptions
{
    MaxDegreeOfParallelism = Environment.ProcessorCount / 2, // 使用一半CPU核心
    CancellationToken = cancellationToken
};

Parallel.For(0, 100, options, i => 
{
    // 计算密集型工作
});

PLINQ (Parallel LINQ)

PLINQ将LINQ查询并行化，只需添加.AsParallel()：

var source = Enumerable.Range(1, 10000);

// 并行查询
var evenNumbers = source.AsParallel()
                       .Where(x => x % 2 == 0)
                       .Select(x => Math.Pow(x, 2))
                       .ToList();

线程安全注意事项

数据并行需要特别注意共享数据的访问：

int sum = 0;
Parallel.For(0, 1000, i => 
{
    Interlocked.Add(ref sum, i); // 线程安全操作
    // 等同于 lock 保护的 sum += i
});

性能考量

数据并行并不总是更快，需要考虑：

• 开销：并行化本身有开销，小数据集可能得不偿失
• 内存局部性：连续内存访问通常更快
• 负载均衡：确保工作均匀分配

graph TD A[开始并行操作] --> B{数据集大小>阈值?} B -->|是| C[分割数据集] B -->|否| D[顺序处理] C --> E[并行处理分区] E --> F[合并结果] F --> G[结束] D --> G

实际应用案例

图像处理是数据并行的典型应用。例如并行应用滤镜：

void ApplyFilterParallel(byte[] pixels, Func<byte, byte> filter)
{
    Parallel.For(0, pixels.Length, i => 
    {
        pixels[i] = filter(pixels[i]);
    });
}

科学计算中矩阵运算也常使用数据并行：

$\left(\begin{array}{ccc}{a}_{11}& \cdots & a_{1n}\\ ⋮& \ddots & ⋮\\ a_{m1}& \cdots & a_{mn}\end{array}\right)+\left(\begin{array}{ccc}{b}_{11}& \cdots & b_{1n}\\ ⋮& \ddots & ⋮\\ b_{m1}& \cdots & b_{mn}\end{array}\right)=\left(\begin{array}{ccc}{a}_{11}+b_{11}& \cdots & a_{1n}+b_{1n}\\ ⋮& \ddots & ⋮\\ a_{m1}+b_{m1}& \cdots & a_{mn}+b_{mn}\end{array}\right)$

并行实现矩阵加法：

double[,] MatrixAddParallel(double[,] a, double[,] b)
{
    int rows = a.GetLength(0);
    int cols = a.GetLength(1);
    double[,] result = new double[rows, cols];
    
    Parallel.For(0, rows, i => 
    {
        for (int j = 0; j < cols; j++)
        {
            result[i, j] = a[i, j] + b[i, j];
        }
    });
    
    return result;
}

最佳实践

1. 测量性能：始终比较并行和顺序版本的性能 2. 避免过度并行化：MaxDegreeOfParallelism通常设为CPU核心数 3. 减少共享状态：尽量使用局部变量 4. 处理异常：使用AggregateException处理并行操作中的异常

总结

C#数据并行提供了强大的工具来处理计算密集型任务：

• 使用Parallel.For/ForEach进行循环并行化
• 使用PLINQ并行化LINQ查询
• 注意线程安全和性能优化
• 适合大规模数据处理和计算密集型应用

正确使用时，数据并行可以显著提高应用程序性能，特别是在多核处理器上。