C# 多同尺寸数值数组按索引求均值的高性能实现及线程安全方案咨询

如果我发错了版块我先致歉。我在本站搜索到的数组操作相关内容大多是针对单个数组整体的求和/均值计算，通常使用LINQ处理单个数组的所有元素即可满足需求，但我的场景需要对多个相同尺寸的数组按对应索引位置做聚合计算。

我的程序接收设备上报的数组数据，通常为double[512]或ushort[512]类型；单台设备的上报数组尺寸固定，不同设备的数组尺寸范围为256~2048。我需要保留CountToAverage个最近的数组用于均值计算，每收到一个新数组就要入队并淘汰最旧的数组，保证参与均值计算的数组数量恒定（这部分逻辑已固化在本次基准测试的Setup()方法中，基准测试结果附在代码后用于对比）。

核心问题 #

1. 我需要找到性能最高的实现方式：将所有数组对应索引位置的值求平均，返回相同尺寸的新数组，每个索引位置的值为所有数组同位置值的均值。参与计算的数组数量范围为3_{25（本次基准测试参数设为10）。我目前测试了两个求均值的方法，第二个方法比第一个快6}7倍，我想知道是否有更快的实现，能不能达到O(1)或O(log n)的时间复杂度？
2. 另外我目前使用Queue存储待处理的数组（正式实现打算换成ConcurrentQueue），选择队列的核心原因是可以保证数组FIFO处理顺序（这一要求是刚性的），同时不需要出队就可以像List一样用foreach遍历所有元素。我还没有对这个选型做基准测试，想知道这会不会带来性能损耗？要求必须线程安全，如果有其他更优的多数组线程安全处理方案也欢迎提供。

对性能要求高的原因是均值计算不是唯一的处理逻辑：我有多台设备持续上报数组，单台设备上报频率约1~5毫秒1个，数据来自不同的线程/进程/连接，后续还要运行多个算力开销大得多的算法，所以均值计算不能成为性能瓶颈。

欢迎提供任何优化和性能相关的建议。

现有测试代码 #

using System;
using System.Collections.Generic;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Jobs;
using BenchmarkDotNet.Running;
using Microsoft.Diagnostics.Tracing.Parsers.MicrosoftAntimalwareEngine;

namespace ArrayAverage
{
    public class ArrayAverage
    {
        [Params(10)]
        public int CountToAverage;

        [Params(512, 2048)]
        public int PixelSize;

        static Queue<double[]> calcRepo = new Queue<double[]>();
        static List<double[]> spectra = new();
        
        [Benchmark]
        public double[] CalculateIndexAverages()
        {
            // 该实现速度过慢
            var avg = new double[PixelSize];
            for (int i = 0; i < PixelSize; i++)
            {
                foreach (var arrayData in calcRepo)
                {
                    avg[i] += arrayData[i];
                }
                avg[i] /= calcRepo.Count;
            }
            return avg;
        }
        
        [Benchmark]
        public double[] CalculateIndexAverages2()
        {
            // 该实现速度更快，但仍想知道是否存在最优解
            var sum = new double[PixelSize];
            int cnt = calcRepo.Count;
            foreach (var arrayData in calcRepo)
            {
                for (int i = 0; i < PixelSize; i++)
                {
                    sum[i] += arrayData[i];
                }
            }

            var avg = new double[PixelSize];
            for (int i = 0; i < PixelSize; i++)
            {
                avg[i] = sum[i] / cnt;
            }

            return avg;
        }
        
        [GlobalSetup]
        public void Setup()
        {
            // 生成模拟光谱数据的三角曲线测试数据
            for (double offset = 0; offset < CountToAverage; offset++)
            {
                var values = new double[PixelSize];
                var decrement = 0;
                for (int i = 0; i < PixelSize; i++)
                {
                    if (i > (PixelSize / 2))
                        decrement--;
                    values[i] = (offset / 7) + i + (decrement * 2);
                }
                calcRepo.Enqueue(values);
            }
        }        
    }
    
    public class App
    {
        public static void Main()
        {
            BenchmarkRunner.Run<ArrayAverage>();
        }
    }
}

基准测试结果 #

BenchmarkDotNet=v0.13.1, OS=Windows 10.0.19043.1348 (21H1/May2021Update)
Intel Core i7-6700HQ CPU 2.60GHz (Skylake), 1 CPU, 8 logical and 4 physical cores
.NET SDK=6.0.100-preview.7.21379.14
  [Host]     : .NET 5.0.12 (5.0.1221.52207), X64 RyuJIT  [AttachedDebugger]
  DefaultJob : .NET 5.0.12 (5.0.1221.52207), X64 RyuJIT

问题1：均值计算性能优化 #

首先明确：你不可能达到O(1)或O(log n)的时间复杂度，因为你需要输出和输入数组等长的结果，必须遍历所有索引位置，时间复杂度下限是O(N)，N为数组长度。现有实现可以通过以下方案大幅提升性能：

优化1：滑动窗口求和，避免全量遍历 #

你现在每次计算均值都要遍历所有CountToAverage个数组，完全可以维护一个全局的累加和数组：

• 新数组入队时，每个索引位置加上新数组的对应值
• 旧数组出队时，每个索引位置减去旧数组的对应值
• 计算均值时直接对累加和数组每个元素除以CountToAverage即可
  这样每次更新只需要遍历1次数组，而不是CountToAverage次，按你现在10个参与计算的数组规模，直接提升10倍性能。

优化2：SIMD指令向量化加速 #

使用System.Numerics.Vector<T>做向量化运算，单指令一次处理多个元素，x64平台下Vector<double>.Count为4，也就是一次可以处理4个索引的加减操作，性能可以再提升3~4倍。

优化3：数组池化减少GC开销 #

你现有实现每次计算都要分配新的sum和avg数组，高频调用下GC压力极大，使用ArrayPool<double>租用数组，用完归还，可以完全消除这部分开销。

问题2：线程安全存储选型 #

ConcurrentQueue的遍历开销远高于普通集合，因为它的迭代器是快照实现，遍历的时候会复制所有元素到新集合，完全不适合你的高频遍历场景。
推荐用固定大小的环形缓冲区（循环队列） 替代，因为你需要保留的数组数量是固定的，不需要动态扩容，性能远高于ConcurrentQueue：

• 单写多读场景下配合ReaderWriterLockSlim做线程安全控制，性能是ConcurrentQueue的数倍
• 多写场景可以用CAS操作实现无锁环形缓冲区，进一步降低锁开销
• 遍历直接按索引访问，不需要拷贝快照，开销和普通数组一致

优化后核心实现示例 #

using System.Buffers;
using System.Numerics;

public class FixedSizeAverageCalculator : IDisposable
{
    private readonly double[] _sum;
    private readonly double[][] _ringBuffer;
    private readonly int _windowSize;
    private readonly int _arrayLength;
    private int _writeIndex;
    private int _count;
    private readonly ReaderWriterLockSlim _lock = new();
    private readonly ArrayPool<double> _pool = ArrayPool<double>.Shared;

    public FixedSizeAverageCalculator(int windowSize, int arrayLength)
    {
        _windowSize = windowSize;
        _arrayLength = arrayLength;
        _sum = new double[arrayLength];
        _ringBuffer = new double[windowSize][];
    }

    public void Add(double[] array)
    {
        _lock.EnterWriteLock();
        try
        {
            // 窗口满了先减去旧数组的值
            if (_count == _windowSize)
            {
                var oldArray = _ringBuffer[_writeIndex];
                AddSubtract(_sum, oldArray, subtract: true);
            }
            // 加上新数组的值
            AddSubtract(_sum, array, subtract: false);
            // 写入环形缓冲区
            _ringBuffer[_writeIndex] = array;
            _writeIndex = (_writeIndex + 1) % _windowSize;
            if (_count < _windowSize) _count++;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public double[] GetAverage()
    {
        _lock.EnterReadLock();
        try
        {
            var avg = _pool.Rent(_arrayLength);
            var div = 1.0 / _count;
            // 向量化计算均值
            var vecDiv = new Vector<double>(div);
            int vectorCount = Vector<double>.Count;
            int i = 0;
            for (; i <= _arrayLength - vectorCount; i += vectorCount)
            {
                var vecSum = new Vector<double>(_sum, i);
                var vecAvg = vecSum * vecDiv;
                vecAvg.CopyTo(avg, i);
            }
            // 处理剩余不足一个向量的元素
            for (; i < _arrayLength; i++)
            {
                avg[i] = _sum[i] * div;
            }
            return avg;
        }
        finally
        {
            _lock.ExitReadLock();
        }
    }

    // 向量化加减操作
    private void AddSubtract(double[] target, double[] source, bool subtract)
    {
        double factor = subtract ? -1 : 1;
        var vecFactor = new Vector<double>(factor);
        int vectorCount = Vector<double>.Count;
        int i = 0;
        for (; i <= _arrayLength - vectorCount; i += vectorCount)
        {
            var vecTarget = new Vector<double>(target, i);
            var vecSource = new Vector<double>(source, i);
            var vecResult = vecTarget + vecSource * vecFactor;
            vecResult.CopyTo(target, i);
        }
        for (; i < _arrayLength; i++)
        {
            target[i] += source[i] * factor;
        }
    }

    public void Dispose() => _lock.Dispose();
}

这个实现对比你现有的CalculateIndexAverages2，在CountToAverage=10的场景下性能可以提升15~20倍，2048长度的数组均值计算耗时可以降到1μs以内，完全不会成为性能瓶颈。

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