使用multiprocessing.Pool递归并行求和报错，寻求解决方案

问题背景 #

想要通过递归并行的方式计算数组所有元素的总和，使用Python的multiprocessing库实现，电脑最多允许3个进程同时执行。运行代码时出现如下错误：

"pool objects cannot be passed between processes or pickled"

报错栈信息：

File "...\parallel_sum.py", line 74, in <module>
    res = parallel_sum.computeSum(a, 0, len(a) - 1)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "...\parallel_sum.py", line 50, in computeSum
    print(result.get())
          ^^^^^^^^^^^^
  File "...\miniconda3\Lib\multiprocessing\pool.py", line 774, in get
    raise self._value
  File "...\miniconda3\Lib\multiprocessing\pool.py", line 540, in _handle_tasks
    put(task)
  File "...\miniconda3\Lib\multiprocessing\connection.py", line 205, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "...\miniconda3\Lib\multiprocessing\reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
  File "...\miniconda3\Lib\multiprocessing\pool.py", line 643, in __reduce__
    raise NotImplementedError(
NotImplementedError: pool objects cannot be passed between processes or pickled

用户原代码：

import multiprocessing as mp
import random

def randList(len):
    l = []
    for _ in range(len):
        l.append(random.randint(1, 100))
    return l

def init_pool(sharedArr_):
    global sharedArr
    sharedArr = sharedArr_


class ParallelSum:
    def solve(self, l, r):
        if l >= r:
            return sharedArr[l]

        m = l + (r - l) // 2

        sumL = self.pool.apply_async(self.solve, args=(l, m)) # compute left part of the array
        sumR = self.pool.apply_async(self.solve, args=(m + 1, r)) # compute right part of the array

        return sumL.get() + sumR.get()

    def computeSum(self, a, l, r):
        sharedArr = mp.Array("i", a)

        self.pool = mp.Pool(initializer=init_pool, initargs=(sharedArr,))

        result = self.pool.apply_async(self.solve, args=(l, r))

        print(result.get()) # Error occurs here
        self.pool.close()
        self.pool.join()


if __name__ == "__main__":
    a = randList(int(5000000)) # generating a list of random integers

    parallel_sum = ParallelSum()

    res = parallel_sum.computeSum(a, 0, len(a) - 1)

    print(res)

错误原因 #

核心问题是：当你把self.solve提交到进程池时，self（ParallelSum类的实例）会被序列化传递给子进程，而self中包含了pool对象——multiprocessing.Pool无法被序列化（pickle）或在进程间传递，这直接触发了报错。

另外，原代码的递归逻辑会无限制提交异步任务，容易导致进程池过载，也没有利用好“最多3个进程”的限制。

修正方案 #

修正思路 #

1. 剥离类实例与进程池的绑定，避免序列化实例时携带无法被pickle的pool对象；
2. 控制递归并行的触发阈值，避免小任务频繁创建进程带来的开销；
3. 固定进程池大小为3，符合系统并发限制；
4. 使用独立函数替代类方法，简化逻辑。

修正后的代码 #

import multiprocessing as mp
import random

def randList(length):
    # 简化列表生成逻辑
    return [random.randint(1, 100) for _ in range(length)]

# 全局共享数组，由进程池初始化
sharedArr = None

def init_pool(shared_arr):
    global sharedArr
    sharedArr = shared_arr

def solve(l, r):
    if l >= r:
        return sharedArr[l]
    
    m = l + (r - l) // 2
    
    # 设置阈值：当区间长度小于1000时，串行计算（避免进程调度开销）
    if r - l < 1000:
        return solve(l, m) + solve(m + 1, r)
    
    # 仅在大任务时使用进程池并行，固定进程数为3
    with mp.Pool(3, initializer=init_pool, initargs=(sharedArr,)) as pool:
        sumL = pool.apply_async(solve, args=(l, m))
        sumR = pool.apply_async(solve, args=(m + 1, r))
        return sumL.get() + sumR.get()

def computeSum(a):
    # 创建进程间共享数组
    shared_arr = mp.Array("i", a)
    # 初始化共享数组并启动进程池计算
    with mp.Pool(3, initializer=init_pool, initargs=(shared_arr,)) as pool:
        result = pool.apply_async(solve, args=(0, len(a) - 1))
        return result.get()

if __name__ == "__main__":
    a = randList(5000000)
    res = computeSum(a)
    print(res)

关键修改点 #

• 移除类结构：将solve改为独立函数，避免序列化包含pool的类实例；
• 并行阈值控制：小任务串行计算，减少进程调度的额外开销；
• 进程池资源管理：用with语句自动管理进程池的创建与销毁，无需手动close()和join()；
• 固定进程数：明确设置进程池大小为3，符合系统并发限制。

内容的提问来源于stack exchange，提问作者Ngô Hiệp