multiprocessing.pool类内调用仅加time.sleep()可正常运行问题

更新：将f函数移到类外部似乎也能让该示例正常运行。

我的实现目标：

• 创建一个可通过进程池处理任务的类
• 创建类实例，启动进程并监控任务执行进度

核心疑问：为什么代码只有在pool.close()调用后执行time.sleep()才能正常工作。

最小复现示例：

import multiprocessing as mp
import time


class MultProc:
    def __init__(self, n: int):
        self.n = n
 
    # 将函数移到类外部定义同样可正常运行
    def f(self, i: int, return_list):
        time.sleep(5)
        return_list.append(i)

    def multiproc(self):
        manager = mp.Manager()
        self.return_list = manager.list()
        pool = mp.Pool()
        results = {}
        for i in range(self.n):
            results[i] = pool.apply_async(self.f, args=(i, self.return_list))
        pool.close()

        time.sleep(0.01) # 仅添加该行代码时程序可正常运行

        self.pool = pool

def main():
    mult = MultProc(n=10)
    mult.multiproc()
    n = 0
    while True:
        n+=1
        print(mult.return_list)
        time.sleep(1)        
        if n > 20:
            break


if __name__ == "__main__":
    main()

这个现象是两个机制叠加导致的：

1. 任务序列化是异步执行的：multiprocessing.Pool的apply_async方法不会在调用瞬间同步完成任务参数的序列化操作，序列化和任务下发是由进程池内部的后台线程异步完成的。你传入的是实例绑定方法self.f，序列化这个方法时需要连带序列化整个self实例。
2. Pool对象不可序列化：你在提交完所有任务、调用pool.close()之后，才执行self.pool = pool把进程池对象绑定到实例属性上，而mp.Pool对象本身是不支持pickle序列化的。

如果没有time.sleep(0.01)：后台序列化线程还没来得及完成self实例的序列化，你就已经把不可序列化的pool属性挂到了self上，序列化过程会直接失败，任务根本不会发送给工作进程。而且这个异常会被进程池的后台线程捕获吞掉，主进程不会收到任何报错，只会看到返回列表一直为空。
如果加了0.01秒的等待：这段时间足够后台线程在你挂载pool属性之前完成所有实例序列化、把任务正常下发给工作进程，后续再修改实例属性也不会影响已经发出去的任务，程序就能正常运行。
至于把f移到类外部就能正常运行，是因为顶层函数序列化时不需要处理self实例，不管你什么时候给实例挂载不可序列化的属性，都不会影响任务下发，自然不需要额外加sleep。

靠固定时长的sleep等待是非常不可靠的，机器性能波动、任务量变化都可能导致等待时长不足。推荐用以下两种稳定方案：

方案1：避免序列化实例对象 #

把任务函数移到类外部，或者用@staticmethod标记为静态方法，让任务函数不需要绑定实例，序列化时就不需要处理整个self对象，从根源上规避问题：

import multiprocessing as mp
import time

def f(i: int, return_list):
    time.sleep(5)
    return_list.append(i)

class MultProc:
    def __init__(self, n: int):
        self.n = n
        self.pool = None
        self.return_list = None

    def multiproc(self):
        manager = mp.Manager()
        self.return_list = manager.list()
        self.pool = mp.Pool()
        results = {}
        for i in range(self.n):
            results[i] = self.pool.apply_async(f, args=(i, self.return_list))
        self.pool.close()

def main():
    mult = MultProc(n=10)
    mult.multiproc()
    n = 0
    while True:
        n+=1
        print(mult.return_list)
        time.sleep(1)        
        if n > 20:
            mult.pool.terminate()
            mult.pool.join()
            break

if __name__ == "__main__":
    main()

方案2：自定义实例序列化逻辑 #

如果一定要保留实例方法，可以通过实现__getstate__方法，指定实例序列化时跳过不可序列化的pool属性，即使提前把pool挂载到实例上，也能正常完成序列化：

import multiprocessing as mp
import time

class MultProc:
    def __init__(self, n: int):
        self.n = n
        self.pool = None
        self.return_list = None

    # 自定义序列化规则，排除不可序列化的pool字段
    def __getstate__(self):
        state = self.__dict__.copy()
        state['pool'] = None
        return state

    def f(self, i: int, return_list):
        time.sleep(5)
        return_list.append(i)

    def multiproc(self):
        manager = mp.Manager()
        self.return_list = manager.list()
        self.pool = mp.Pool()
        results = {}
        for i in range(self.n):
            results[i] = self.pool.apply_async(self.f, args=(i, self.return_list))
        self.pool.close()

def main():
    mult = MultProc(n=10)
    mult.multiproc()
    n = 0
    while True:
        n+=1
        print(mult.return_list)
        time.sleep(1)        
        if n > 20:
            mult.pool.terminate()
            mult.pool.join()
            break

if __name__ == "__main__":
    main()

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