C#中AsParallel与Parallel如何降低结果合并频率？

问题场景 #

现有一个嵌套循环用于并行归约（MapReduce），但每次迭代都直接更新全局字典，导致严重的缓存竞争和性能浪费：

// 字典规模较小，易引发竞争
Dictionary<SumDataCategory, SumData> globalResult;
// 待并行化的循环
for (int i = 0; i < N; i++) {
  for (int j = 0; j < N; j++) {
    for (int k = 0; k < N; k++) {
      // 根据i,j,k获取输入数据
      InputData[] inputs = fetchInputData(i, j, k);
      // 执行独立计算
      SumData sum1 = SumData.Sum(inputs);
      // 最终归约操作 <- 待优化
      globalResult[sum1.Category].Sum(sum1);
    }
  }
}

目标与约束 #

• 目标：最小改动优化归约性能，无需重写整个循环/处理逻辑
• 工作负载约束：
  • 非完全规整，不能假设静态分区
  • 非高度动态，无需动态生成作业

现有问题分析 #

• 每次迭代直接更新全局状态，不仅浪费计算资源，还会引发严重的缓存竞争（迭代/输入量远多于CPU核心数）
• 手动分区实现复杂，C#的AsParallel/Parallel是纯库实现（无编译器辅助，不同于OpenMP），大部分逻辑需自行处理
• 按块归约到全局状态依然存在浪费，且并发合并的复杂度高于最终串行/并行合并

可行解决方案（对标OpenMP线程本地归约逻辑） #

针对C#的Parallel类，我们可以利用ThreadLocal<T>实现线程本地归约，最后合并所有线程的本地结果，完美匹配OpenMP的思路：

步骤1：定义线程本地的归约容器 #

使用ThreadLocal<Dictionary<SumDataCategory, SumData>>让每个线程维护自己的本地字典，避免全局竞争：

// 初始化线程本地字典，每个线程首次访问时创建新的字典实例
// 假设SumData有Clone方法复制初始状态，确保线程间数据独立
var threadLocalResults = new ThreadLocal<Dictionary<SumDataCategory, SumData>>(() => 
    globalResult.ToDictionary(kv => kv.Key, kv => kv.Value.Clone()));

步骤2：并行化循环并执行线程本地归约 #

用Parallel.For重构嵌套循环，在每个迭代中更新线程本地字典：

// 并行化三层循环
Parallel.For(0, N, i => {
    Parallel.For(0, N, j => {
        Parallel.For(0, N, k => {
            InputData[] inputs = fetchInputData(i, j, k);
            SumData sum1 = SumData.Sum(inputs);
            
            // 更新线程本地字典，无全局竞争
            var localDict = threadLocalResults.Value;
            localDict[sum1.Category].Sum(sum1);
        });
    });
});

注：如果三层循环的总迭代数可以展开为单一集合，也可以用Enumerable.Range结合AsParallel简化代码：

Enumerable.Range(0, N*N*N)
    .AsParallel()
    .WithDegreeOfParallelism(Environment.ProcessorCount)
    .ForEach(idx => {
        int i = idx / (N*N);
        int j = (idx / N) % N;
        int k = idx % N;
        // 后续逻辑同上，更新线程本地字典
    });

步骤3：合并所有线程本地结果到全局字典 #

循环结束后，串行合并所有线程的本地字典到全局结果（若需更高效率，可改用分段锁或ConcurrentDictionary实现并行合并）：

lock (globalResult)
{
    foreach (var localDict in threadLocalResults.Values)
    {
        foreach (var kv in localDict)
        {
            globalResult[kv.Key].Sum(kv.Value);
        }
    }
}
// 释放线程本地资源
threadLocalResults.Dispose();

关键细节说明 #

1. SumData的Clone与合并逻辑：确保每个线程的本地字典持有独立的SumData实例，避免线程间意外共享；合并时要实现正确的累加逻辑（比如SumData的Sum方法支持合并另一个SumData实例）
2. ThreadLocal的初始化：必须保证每个线程首次访问时拿到全局结果的副本，而非引用，否则仍会出现竞争
3. 并行度控制：可通过WithDegreeOfParallelism指定并行线程数，避免过度调度

替代方案：使用PLINQ的Aggregate方法 #

如果可以将迭代转换为集合，PLINQ的Aggregate方法原生支持分阶段归约（先线程本地归约，再全局合并），代码更简洁：

var finalResult = Enumerable.Range(0, N*N*N)
    .AsParallel()
    .Aggregate(
        // 初始化线程本地归约容器
        () => globalResult.ToDictionary(kv => kv.Key, kv => kv.Value.Clone()),
        // 线程本地归约逻辑
        (localDict, idx) => {
            int i = idx / (N*N);
            int j = (idx / N) % N;
            int k = idx % N;
            InputData[] inputs = fetchInputData(i, j, k);
            SumData sum1 = SumData.Sum(inputs);
            localDict[sum1.Category].Sum(sum1);
            return localDict;
        },
        // 合并两个本地容器
        (dict1, dict2) => {
            foreach (var kv in dict2)
            {
                dict1[kv.Key].Sum(kv.Value);
            }
            return dict1;
        },
        // 最终结果处理
        finalDict => finalDict
    );

// 将结果同步回全局字典（若需要）
lock (globalResult)
{
    foreach (var kv in finalResult)
    {
        globalResult[kv.Key] = kv.Value;
    }
}

这个方案完全由PLINQ处理线程本地归约和合并逻辑，无需手动管理ThreadLocal，改动量更小。

内容的提问来源于stack exchange，提问作者user2771324