矩形法则积分代码效率优化及多进程实现技术咨询

针对你用矩形法则计算f(x)=cos(cos(cos(cos(cos(x²)))))在[0,3]区间积分的场景，从易到难给你几个优化方向，优先从向量化和JIT编译入手，多进程是备选方案：

1. 用NumPy向量化替代Python循环（最见效的优化） #

Python原生for循环是效率瓶颈的核心原因，用NumPy的向量化操作可以把整个求和过程变成数组运算，完全避开循环，效率能提升几个数量级。

示例代码：

import numpy as np

def f(x):
    x_sq = x ** 2
    res = np.cos(x_sq)
    # 连续套4层cos
    for _ in range(4):
        res = np.cos(res)
    return res

def rectangle_integral(a, b, n):
    dx = (b - a) / n
    # 中点矩形法则，生成所有采样点
    x = np.linspace(a + dx/2, b - dx/2, n)
    return dx * np.sum(f(x))

# 调用示例（1000万采样点）
result = rectangle_integral(0, 3, 10_000_000)
print(result)

2. Numba JIT编译加速循环 #

如果不想用NumPy，或者函数逻辑复杂没法完全向量化，用Numba把求和函数编译成机器码，能让Python循环的效率接近C语言。

示例代码：

from numba import njit
import math

@njit(fastmath=True)
def f(x):
    x_sq = x * x
    res = math.cos(x_sq)
    for _ in range(4):
        res = math.cos(res)
    return res

@njit(fastmath=True)
def rectangle_integral(a, b, n):
    dx = (b - a) / n
    total = 0.0
    for i in range(n):
        # 中点矩形采样
        x = a + dx/2 + i * dx
        total += f(x)
    return dx * total

# 调用示例
result = rectangle_integral(0, 3, 10_000_000)
print(result)

第一次调用会触发编译，之后的调用都是纯机器码执行，速度极快。

3. 多进程并行计算 #

如果你的CPU有多核心，且采样点数量极大，可以把采样区间拆分成多个块，用多进程并行计算每个块的求和，最后合并结果。注意：多进程有进程间通信开销，只有当n足够大时才划算。

示例代码用concurrent.futures实现：

import numpy as np
from concurrent.futures import ProcessPoolExecutor

def f(x):
    x_sq = x ** 2
    res = np.cos(x_sq)
    for _ in range(4):
        res = np.cos(res)
    return res

def compute_chunk(args):
    a, b, n_chunk = args
    dx = (b - a) / n_chunk
    x = np.linspace(a + dx/2, b - dx/2, n_chunk)
    return np.sum(f(x))

def parallel_rectangle_integral(a, b, n, num_workers=None):
    dx = (b - a) / n
    # 默认按CPU核心数拆分块
    num_workers = num_workers or 4
    chunk_size = n // num_workers
    chunks = []
    current_a = a
    
    for i in range(num_workers):
        if i == num_workers - 1:
            current_b = b
        else:
            current_b = current_a + chunk_size * dx
        chunks.append((current_a, current_b, chunk_size))
        current_a = current_b
    
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        sums = list(executor.map(compute_chunk, chunks))
    
    return dx * sum(sums)

# 调用示例
result = parallel_rectangle_integral(0, 3, 10_000_000, num_workers=4)
print(result)

额外小优化 #

• 减少重复计算：你的函数里x²只需要计算一次，然后依次套cos，不要每次计算cos时重复计算x²，上面的示例已经做了这点。
• 选择中点矩形法则：它的精度比左/右点矩形法则更高，相同精度要求下可以用更少的采样点，间接提升效率。

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