如何阻止concurrent.futures重复初始化pygame库

问题背景 #

在学习使用multiprocessing并结合Pygame的项目中，尝试用concurrent.futures实现多进程时，发现Pygame被频繁重复初始化，即便子进程执行的函数完全不需要Pygame。以下是简化后的测试代码及不同Python版本的运行结果。

测试代码 #

import concurrent.futures
import time
import pygame

def do_something(seconds):
    print('zzz', seconds)
    time.sleep(seconds)
    return 'Wake up'

if __name__ == '__main__':
    start = time.perf_counter()

    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = [executor.submit(do_something, 1) for _ in range(3)]

        for f in concurrent.futures.as_completed(results):
            print(f.result())

    finish = time.perf_counter()
    print(f"Finished in {round(finish-start, 2)} seconds.")

不同Python版本的运行结果 #

Python 3.8.13 输出 #

python multiprocessing_test.py
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
zzz 1
zzz 1
zzz 1
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.8.13)
Hello from the pygame community. https://www.pygame.org/contribute.html
Wake up
Wake up
Wake up
Finished in 1.54 seconds.

仅启动3个进程，但Pygame的初始化提示被打印了12次。

Python 3.9.16 输出 #

python multiprocessing_test.py
pygame 2.1.2 (SDL 2.0.18, Python 3.9.16)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.9.16)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.9.16)
Hello from the pygame community. https://www.pygame.org/contribute.html
pygame 2.1.2 (SDL 2.0.18, Python 3.9.16)
Hello from the pygame community. https://www.pygame.org/contribute.html
zzz 1
zzz 1
zzz 1
Wake up
Wake up
Wake up
Finished in 1.18 seconds.

Pygame初始化提示仅被打印4次（主进程1次，3个子进程各1次）。

问题分析与解决方案 #

原因解析 #

1. 多进程启动机制差异：
   • Python 3.8及之前的ProcessPoolExecutor在Unix系统下默认使用fork方式创建进程，fork会复制父进程的整个内存空间，包括已导入的Pygame模块。但子进程启动时会重新执行模块级代码，导致Pygame的导入提示被重复打印，甚至可能引发实际的重复初始化问题。
   • Python 3.9之后，ProcessPoolExecutor默认改用spawn方式，spawn会启动全新的Python解释器，仅执行__main__块中的代码，模块级导入只会在子进程启动时执行一次，因此初始化提示次数减少。
2. Pygame的导入特性：Pygame在被导入时会自动打印版本和社区提示，这是模块级的副作用，每次导入（包括子进程中的导入）都会触发该输出。

解决方法 #

1. 延迟导入Pygame

将Pygame的导入放到if __name__ == '__main__'块内部，确保只有主进程会导入Pygame，子进程不会执行这段导入代码：

import concurrent.futures
import time

def do_something(seconds):
    print('zzz', seconds)
    time.sleep(seconds)
    return 'Wake up'

if __name__ == '__main__':
    import pygame  # 仅主进程导入Pygame
    start = time.perf_counter()

    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = [executor.submit(do_something, 1) for _ in range(3)]

        for f in concurrent.futures.as_completed(results):
            print(f.result())

    finish = time.perf_counter()
    print(f"Finished in {round(finish-start, 2)} seconds.")

2. 指定进程启动方式为spawn

即使在Python 3.8中，也可以手动设置ProcessPoolExecutor使用spawn方式启动进程，避免子进程继承父进程的Pygame状态：

import concurrent.futures
import time
import multiprocessing
import pygame

def do_something(seconds):
    print('zzz', seconds)
    time.sleep(seconds)
    return 'Wake up'

if __name__ == '__main__':
    start = time.perf_counter()

    # 强制使用spawn启动子进程
    with concurrent.futures.ProcessPoolExecutor(mp_context=multiprocessing.get_context('spawn')) as executor:
        results = [executor.submit(do_something, 1) for _ in range(3)]

        for f in concurrent.futures.as_completed(results):
            print(f.result())

    finish = time.perf_counter()
    print(f"Finished in {round(finish-start, 2)} seconds.")

3. 关闭Pygame的导入提示

如果只是不想看到打印的提示信息，可以在导入Pygame前设置环境变量屏蔽输出：

import os
os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "1"
import pygame

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