为何ForEach与Parallel.For性能远低于for循环？

本文讨论的场景是：对两个数组执行按位与操作，统计结果中1位的总数量。以下是三种实现方式、测试结果及性能分析：

三种实现代码 #

普通for循环 #

//For Approach
for (int i = 0; i < n; i++) { 
    countOnes += Int32.PopCount( x[i] & y[i]); 
}

ForEach+Zip实现 #

//Foreach approach
foreach( var (a,b) in x.Zip(y) ) { 
    countOnesZip += Int32.PopCount( a & b); 
}

Parallel.For实现 #

//Parallel approach
Parallel.For(0, n, i =>
{
    Interlocked.Add(ref countOnesPar, Int32.PopCount(x[i] & y[i]));
});

测试结果 #

Processed 100,000,000 bits
Elapsed time (For): 11ms Count: 24,216,440
Elapsed time (ForEach): 96ms Count: 24,216,440
Elapsed time (Parallel.For): 107ms Count: 24,216,440

性能差异原因 #

ForEach慢的核心原因 #

Zip方法会创建额外的枚举器对象，每次迭代都要处理枚举的状态维护和对象分配；而普通for循环直接通过索引访问数组——数组的索引访问是CLR中效率最高的集合访问方式之一，没有任何额外的对象开销和枚举逻辑，自然速度快很多。

Parallel.For慢的核心原因 #

问题出在Interlocked.Add这个原子操作上。每次迭代都调用它更新共享的countOnesPar变量，原子操作需要CPU层面的缓存一致性同步（比如锁总线、MESI协议交互），会带来极大的性能开销。多个线程频繁竞争同一个共享变量时，大部分时间都在等待同步，完全抵消了并行计算的优势，甚至比单线程还慢。

优化后的并行实现 #

分块Task方案 #

把数组按CPU核心数分成多个块，每个Task单独计算块内的局部计数，最后汇总所有局部结果，彻底避免频繁的原子操作：

//Tasks版本
var split = new int[Environment.ProcessorCount];
var tasks = new Task[split.Length];
var take = (int)Math.Round((double)x.Length / tasks.Length, MidpointRounding.AwayFromZero);

for (var i = 0; i < tasks.Length; i++)
{
    var taskNumber = i;
    tasks[i] = Task.Run(() =>
    {
        var localCount = 0; 
        for (var index = taskNumber * take; index < (taskNumber + 1) * take && index < x.Length; index++)
        {
            localCount += int.PopCount(x[index] & y[index]);
        }

        split[taskNumber] = localCount;
    });
}

await Task.WhenAll(tasks);
countOnesPar = split.Sum();

Partitioner并行方案 #

使用Partitioner将数据划分为连续的范围，每个范围单独计算局部计数，最后只做一次原子操作汇总局部结果，大幅减少原子操作的次数：

//Partitioner版本
Partitioner<Tuple<int, int>> partitioner = Partitioner.Create(0, n);
Parallel.ForEach(partitioner, range =>
{
    int startIndex = range.Item1;
    int endIndex = range.Item2;
    int localCountOnes = 0;

    for (int i = startIndex; i < endIndex; i++)
    {
        localCountOnes += Int32.PopCount(x[i] & y[i]);
    }

    Interlocked.Add(ref countOnesPar, localCountOnes);
});

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