C#中快速统计两个二维数组对应位置相等元素数量的方法

我有两个尺寸相同的二维0-1数组（最大20×20），需要统计对应位置元素相等的数量。这个函数会被频繁调用，所以想找更高效的实现方式（包括unsafe/汇编）。试过按行并行求和，但性能反而更差。

现有实现代码 #

普通循环版本 #

public int Compare(byte[,] a, byte[,] b)
{
    int score = 0;

    if (a.Length != b.Length)
        return -1;

    for (int y = 0; y < a.GetLength(1); y++)
    {
        for (int x = 0; x < a.GetLength(0); x++)
        {
            if (a[x, y] == b[x, y])
                score++;
        }
    }

    return score;
}

并行版本 #

public int CompareParallel(byte[,] a, byte[,] b)
{
    int[] yScore = new int[a.GetLength(1)];

    if (a.Length != b.Length)
        return -1;

    ParallelLoopResult result = Parallel.For(0, a.GetLength(1), y => {
        for (int x = 0; x < a.GetLength(0); x++)
        {
            if (a[x, y] == b[x, y])
                yScore[y]++;
        }
    });

    return yScore.Sum();
}

测试代码 #

int score;
int iterations = 100000000;
Stopwatch s = new Stopwatch();
byte[,] a = new byte[4, 2] { { 1, 0 }, { 1, 1 }, { 0, 0 }, { 1, 1 } };
byte[,] b = new byte[4, 2] { { 1, 0 }, { 0, 1 }, { 0, 0 }, { 1, 1 } };

s.Start();
    
for(int i = 0; i < iterations; i++)
    score = Compare(a, b);

Console.WriteLine($"TEST1 - Elapsed: {s.Elapsed.Seconds} seconds");

s.Restart();

for (int i = 0; i < iterations; i++)
    score = CompareParallel(a, b);

Console.WriteLine($"TEST2 - Elapsed: {s.Elapsed.Seconds} seconds");

测试结果 #

TEST1 - Elapsed: 3 seconds
TEST2 - Elapsed: <TOO MUCH>

优化方案 #

1. 优化遍历顺序，提升缓存命中率 #

.NET的二维数组是列优先存储的，原代码外层循环遍历列（y）、内层遍历行（x）会导致内存访问不连续，缓存命中率极低。改成行优先遍历（外层行x，内层列y），能大幅利用CPU缓存：

public int CompareOptimized(byte[,] a, byte[,] b)
{
    int score = 0;
    if (a.Length != b.Length) return -1;

    int rows = a.GetLength(0);
    int cols = a.GetLength(1);

    for (int x = 0; x < rows; x++)
    {
        for (int y = 0; y < cols; y++)
        {
            if (a[x, y] == b[x, y])
                score++;
        }
    }
    return score;
}

2. Unsafe内存直接操作，减少边界检查 #

把二维数组视为连续字节块，跳过.NET数组的索引边界检查，同时按int批量处理字节，利用位运算统计相同元素数量：

public unsafe int CompareUnsafe(byte[,] a, byte[,] b)
{
    if (a.Length != b.Length) return -1;

    int count = 0;
    int length = a.Length;

    fixed (byte* pA = a, pB = b)
    {
        byte* ptrA = pA;
        byte* ptrB = pB;

        // 按int批量处理，减少循环次数
        int intBatchCount = length / 4;
        for (int i = 0; i < intBatchCount; i++)
        {
            int valA = *(int*)ptrA;
            int valB = *(int*)ptrB;
            // 异或后0的位数就是相同元素的数量
            int xorResult = valA ^ valB;
            count += 32 - BitOperations.PopCount((uint)xorResult);
            ptrA += 4;
            ptrB += 4;
        }

        // 处理剩余不足4个的字节
        int remainder = length % 4;
        for (int i = 0; i < remainder; i++)
        {
            if (*ptrA == *ptrB) count++;
            ptrA++;
            ptrB++;
        }
    }
    return count;
}

3. SIMD指令批量处理（.NET 5+） #

利用.NET的Vector<T>类实现单指令多数据并行计算，适合20×20=400字节的规模，一次性处理多个字节：

public int CompareSIMD(byte[,] a, byte[,] b)
{
    if (a.Length != b.Length) return -1;

    int count = 0;
    int length = a.Length;
    int vectorSize = Vector<byte>.Count;
    int vectorBatchCount = length / vectorSize;

    Span<byte> spanA = a.AsSpan();
    Span<byte> spanB = b.AsSpan();

    for (int i = 0; i < vectorBatchCount; i++)
    {
        Vector<byte> vecA = new Vector<byte>(spanA.Slice(i * vectorSize));
        Vector<byte> vecB = new Vector<byte>(spanB.Slice(i * vectorSize));
        // 相等的字节会被标记为0xFF，用Dot统计数量
        Vector<byte> equals = Vector.Equals(vecA, vecB);
        count += Vector.Dot(equals, Vector<byte>.One) / 255;
    }

    // 处理剩余字节
    for (int i = vectorBatchCount * vectorSize; i < length; i++)
    {
        if (spanA[i] == spanB[i]) count++;
    }

    return count;
}

并行版本性能差的原因 #

数组规模太小（最大400个元素），并行创建线程、线程上下文切换的开销远大于计算本身。只有当数组规模达到数千元素以上时，并行计算才有性能收益。

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