使用multiprocess库apply_async时未实现异步执行的问题排查

我正在测试multiprocess库的apply_async方法，但发现每次调用都在MainProcess中执行，并未真正实现异步，需等待前一个函数完成才会执行下一个。我使用的是Windows系统与Python 3.8，采用spawn方式创建进程。

原代码： #

import os
import time
from multiprocessing import Pool, cpu_count, current_process
from threading import current_thread


def go_to_sleep():
    pid = os.getpid()
    thread_name = current_thread().name
    process_name = current_process().name
    print(f"{pid} Process {process_name} and {thread_name} going to sleep")
    time.sleep(5)


def apply_async():
    pool = Pool(processes=cpu_count())
    print(f"Number of procesess {len(pool._pool)}")
    for i in range(20):
        pool.apply_async(go_to_sleep())
    pool.close()
    pool.join()


def main():
    apply_async()


if __name__ == "__main__":
    start_time = time.perf_counter()
    main()
    end_time = time.perf_counter()
    print(f"Elapsed run time: {end_time - start_time} seconds.")

输出结果： #

Number of procesess 8
26776 Process MainProcess and MainThread going to sleep
26776 Process MainProcess and MainThread going to sleep
26776 Process MainProcess and MainThread going to sleep

问题原因与解决方法 #

• 核心错误：调用pool.apply_async时，你直接执行了go_to_sleep()，这会让函数在主进程中同步执行，而非提交给进程池异步处理。apply_async需要接收的是函数对象，不是函数执行后的返回值。
• 修复方式：将pool.apply_async(go_to_sleep())修改为pool.apply_async(go_to_sleep)，去掉括号，传递函数本身即可。
• 传参说明：如果需要给目标函数传递参数，使用args参数指定，例如pool.apply_async(go_to_sleep, args=(param1, param2))。

修正后的完整代码： #

import os
import time
from multiprocessing import Pool, cpu_count, current_process
from threading import current_thread


def go_to_sleep():
    pid = os.getpid()
    thread_name = current_thread().name
    process_name = current_process().name
    print(f"{pid} Process {process_name} and {thread_name} going to sleep")
    time.sleep(5)


def apply_async():
    pool = Pool(processes=cpu_count())
    print(f"Number of procesess {len(pool._pool)}")
    for i in range(20):
        # 去掉函数后的括号，传递函数对象
        pool.apply_async(go_to_sleep)
    pool.close()
    pool.join()


def main():
    apply_async()


if __name__ == "__main__":
    start_time = time.perf_counter()
    main()
    end_time = time.perf_counter()
    print(f"Elapsed run time: {end_time - start_time} seconds.")

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