信号处理中Numpy数组按频率步的并行计算实现方案咨询

由于各频率对应的矩阵计算完全独立，以下几种并行方案可以大幅提升处理效率：

方案1：使用标准库concurrent.futures.ProcessPoolExecutor #

通过进程池绕开Python GIL限制，充分利用多核CPU，适合CPU密集型的耗时任务：

import numpy as np
from concurrent.futures import ProcessPoolExecutor

# 假设你的dothing函数已定义
def dothing(matrix):
    # 替换为你的实际耗时处理逻辑
    return processed_matrix

# 辅助函数：封装参数并返回索引与处理结果
def process_single_task(args):
    index, _, matrix_block = args
    return index, dothing(matrix_block)

# 准备任务列表：按频率顺序打包索引、频率值、对应矩阵块
tasks = [
    (idx, freq, A[:, :, idx]) 
    for idx, freq in enumerate(frequencies)
]

# 预分配结果数组
B = np.zeros_like(A)

# 启动进程池并行处理
with ProcessPoolExecutor() as executor:
    # 按任务顺序获取结果，直接赋值到对应位置
    for idx, processed_matrix in executor.map(process_single_task, tasks):
        B[:, :, idx] = processed_matrix

方案2：使用joblib简化并行代码 #

joblib专为数值计算场景优化，对Numpy数组的传递效率更高，代码更简洁：

import numpy as np
from joblib import Parallel, delayed

# 假设你的dothing函数已定义
def dothing(matrix):
    # 替换为你的实际耗时处理逻辑
    return processed_matrix

# 并行处理所有频率矩阵，n_jobs=-1表示使用全部CPU核心
processed_list = Parallel(n_jobs=-1, verbose=10)(
    delayed(dothing)(A[:, :, idx]) 
    for idx, _ in enumerate(frequencies)
)

# 将结果列表转换为3D数组，与原数组形状一致
B = np.stack(processed_list, axis=2)

方案3：使用numba编译加速+并行 #

如果dothing是纯数值计算逻辑，可通过numba将Python代码编译为机器码，同时开启并行循环，效率比进程池更高：

import numpy as np
from numba import njit, prange

# 用numba装饰器开启编译与并行支持
@njit(parallel=True)
def dothing_numba(matrix):
    result = np.zeros_like(matrix)
    # 用prange替代range实现并行循环
    for i in prange(matrix.shape[0]):
        for j in prange(matrix.shape[1]):
            # 替换为你的实际计算逻辑
            result[i, j] = matrix[i, j] ** 2
    return result

# 预分配结果数组并并行处理每个频率矩阵
B = np.zeros_like(A)
for idx in prange(A.shape[2]):
    B[:, :, idx] = dothing_numba(A[:, :, idx])

注意事项 #

• 若dothing依赖全局变量或外部资源，需确保子进程能正常访问（进程池模式下全局变量会被拷贝）；
• 进程池存在启动和数据传递开销，但你的任务耗时极长，完全可以抵消这些开销；
• 内存方面，单个81×81矩阵内存占用极小，并行处理不会有明显压力。

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