如何从四通道设备采集数组中快速分离各通道数据

我有一个从四通道设备采集得到的int类型数组，设备每秒会向数组中填充约200000条数据，数据排列顺序如下：

int[] data = new int[]{1,2,3,4,1,2,3,4,1,2,3,4,1,2,3,4,...};
//data[0] 来自通道0
//data[1] 来自通道1 
//data[2] 来自通道2 
//data[3] 来自通道3
//data[4] 来自通道0
//data[5] 来自通道1 
//data[6] 来自通道2 
//data[7] 来自通道3
// 以此类推...

每秒我都需要将各通道的数据分离出来进行处理，目前使用以下代码实现拆分：

double[] ai1 = new double[data.Length/4];
double[] ai2 = new double[data.Length/4];
double[] ai3 = new double[data.Length/4];
double[] ai4 = new double[data.Length/4];
for (int i = 0; i < data.Length/4; i++)
{
    ai1[i] = data[i * 4 + 0];
    ai2[i] = data[i * 4 + 1];
    ai3[i] = data[i * 4 + 2];
    ai4[i] = data[i * 4 + 3];
}

请问是否存在更快的方法将该数组拆分为四个子数组？

更新测试结果 #

我尝试用Parallel.For实现拆分，在我的电脑上耗时约0.02ms；而普通for循环仅耗时约0.001ms！以下是Parallel.For的实现代码：

double[] ai1 = new double[data.Length / 4];
double[] ai2 = new double[data.Length / 4];
double[] ai3 = new double[data.Length / 4];
double[] ai4 = new double[data.Length / 4];
Parallel.For(0, data.Length / 4, i =>
{
    ai1[i] = data[i * 4 + 0];
    ai2[i] = data[i * 4 + 1];
    ai3[i] = data[i * 4 + 2];
    ai4[i] = data[i * 4 + 3];
});

可行的优化方案 #

• 保留普通for循环，优化内存访问
  普通for循环的性能已经非常出色，因为它按顺序访问原数组的连续内存区域，CPU缓存命中率极高，这是Parallel.For无法比拟的——线程调度的额外开销在单批次5万条/通道的场景下完全抵消了并行的收益，甚至拖慢速度。

• 避免不必要的类型转换
  如果后续处理允许使用int类型，直接创建int类型的子数组，省去int转double的类型转换开销，能进一步提升效率：

  int count = data.Length / 4;
  int[] ai1 = new int[count];
  int[] ai2 = new int[count];
  int[] ai3 = new int[count];
  int[] ai4 = new int[count];
  for (int i = 0; i < count; i++)
  {
      int baseIdx = i * 4;
      ai1[i] = data[baseIdx];
      ai2[i] = data[baseIdx + 1];
      ai3[i] = data[baseIdx + 2];
      ai4[i] = data[baseIdx + 3];
  }
  

• 利用Span提升内存效率
  使用.NET的Span<T>类型可以避免堆内存分配的开销（如果后续处理支持直接使用Span而非数组），或者提升内存访问的连续性：

  int count = data.Length / 4;
  // 使用栈分配的Span，避免堆内存开销
  Span<int> ai1 = stackalloc int[count];
  Span<int> ai2 = stackalloc int[count];
  Span<int> ai3 = stackalloc int[count];
  Span<int> ai4 = stackalloc int[count];
  for (int i = 0; i < count; i++)
  {
      int baseIdx = i * 4;
      ai1[i] = data[baseIdx];
      ai2[i] = data[baseIdx + 1];
      ai3[i] = data[baseIdx + 2];
      ai4[i] = data[baseIdx + 3];
  }
  // 后续直接通过Span进行数据处理
  

• 手动循环展开（可选）
  对于固定4通道的场景，可以手动展开循环，减少循环迭代次数，不过现代编译器通常会自动做类似优化，手动修改的收益有限，示例如下：

  int count = data.Length / 4;
  int[] ai1 = new int[count];
  int[] ai2 = new int[count];
  int[] ai3 = new int[count];
  int[] ai4 = new int[count];
  int i = 0;
  // 每次处理2组数据，减少循环次数
  for (; i < count - 1; i += 2)
  {
      int baseIdx1 = i * 4;
      ai1[i] = data[baseIdx1];
      ai2[i] = data[baseIdx1 + 1];
      ai3[i] = data[baseIdx1 + 2];
      ai4[i] = data[baseIdx1 + 3];
      
      int baseIdx2 = (i + 1) * 4;
      ai1[i + 1] = data[baseIdx2];
      ai2[i + 1] = data[baseIdx2 + 1];
      ai3[i + 1] = data[baseIdx2 + 2];
      ai4[i + 1] = data[baseIdx2 + 3];
  }
  // 处理剩余的单个组
  if (i < count)
  {
      int baseIdx = i * 4;
      ai1[i] = data[baseIdx];
      ai2[i] = data[baseIdx + 1];
      ai3[i] = data[baseIdx + 2];
      ai4[i] = data[baseIdx + 3];
  }
  

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