能否控制Python多进程入口点？多进程重复导入及IPC疑问

在Windows 10（Python 3.9.13）环境下，我在entry.py中定义run_entry作为多进程入口，使用ProcessPoolExecutor创建子进程时，发现新进程会重复导入main.py，而非仅加载entry.py。同时希望通过共享内存Queue实现进程间通信，不想用subprocess但也想了解subprocess中如何实现类似Queue的通信。

代码示例与运行输出 #

main.py #

from entry import call_multi_process

print(f'i am IMPORTED MAIN : {__name__} <-- WHY THIS LOADED MULTIPLE TIMES?')

def main():
    call_multi_process()

if __name__ == '__main__':
    print('i am the GENESIS OF MAIN')
    main()
    print('i am the ENDGAME OF MAIN')

entry.py #

import time
from concurrent.futures import ProcessPoolExecutor

import psutil

print(f'i am IMPORTED AS MULTIPROCESS ENTRY OWNER : {__name__}')

def run_entry(num: int):
    name = psutil.Process().name()

    print(f'I AM MULTIPROCESS BODY / {__name__} / {num}:{name} {psutil.Process().pid}')
    time.sleep(1)

def call_multi_process():
    # I WANT PURE NEWLY CREATED PROCESS ONLY WITH IMPORTING THIS PY FILE WITHOUT MAIN
    with ProcessPoolExecutor(initargs=('name', 'test')) as pool:
        for i in range(3):
            pool.submit(run_entry, i)
            # time.sleep(1)

运行输出 #

i am IMPORTED AS MULTIPROCESS ENTRY OWNER : pyutil.tests.new_process_entry
i am IMPORTED MAIN : __main__ <-- WHY THIS LOADED MULTIPLE TIMES?
i am the GENESIS OF MAIN
i am IMPORTED AS MULTIPROCESS ENTRY OWNER : pyutil.tests.new_process_entry
**i am IMPORTED MAIN : __mp_main__ <-- WHY THIS LOADED MULTIPLE TIMES?**
I AM MULTIPROCESS BODY / pyutil.tests.new_process_entry / 0:python.exe 30596
i am IMPORTED AS MULTIPROCESS ENTRY OWNER : pyutil.tests.new_process_entry
**i am IMPORTED MAIN : __mp_main__ <-- WHY THIS LOADED MULTIPLE TIMES?**
I AM MULTIPROCESS BODY / pyutil.tests.new_process_entry / 1:python.exe 28200
i am IMPORTED AS MULTIPROCESS ENTRY OWNER : pyutil.tests.new_process_entry
**i am IMPORTED MAIN : __mp_main__ <-- WHY THIS LOADED MULTIPLE TIMES?**
I AM MULTIPROCESS BODY / pyutil.tests.new_process_entry / 2:python.exe 6336
i am the ENDGAME OF MAIN

问题解答 #

1. 为什么main.py会被多次加载？ #

Windows系统下，Python多进程默认使用spawn启动方式：子进程会重新启动一个Python解释器，然后导入父进程的主模块（即你运行的main.py），以此完成函数、数据的序列化与反序列化。子进程中main.py的__name__会被设为__mp_main__，因此模块级的代码（比如你写的print语句）会被执行，但if __name__ == '__main__'块内的代码不会运行。

解决办法

将main.py中的模块级代码移到if __name__ == '__main__'块内，避免子进程执行不必要的代码：

from entry import call_multi_process

def main():
    call_multi_process()

if __name__ == '__main__':
    print('i am the GENESIS OF MAIN')
    print(f'i am IMPORTED MAIN : {__name__}')
    main()
    print('i am the ENDGAME OF MAIN')

2. 用ProcessPoolExecutor实现共享内存Queue通信 #

可以借助multiprocessing.Queue实现进程间共享内存通信，通过initializer参数将Queue传递给所有子进程：

修改后的entry.py：

import time
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Queue, get_context

import psutil

print(f'i am IMPORTED AS MULTIPROCESS ENTRY OWNER : {__name__}')

# 子进程全局变量，用于存储共享Queue
worker_queue = None

def init_worker(q):
    """初始化子进程，传入共享Queue"""
    global worker_queue
    worker_queue = q

def run_entry(num: int):
    name = psutil.Process().name()
    print(f'I AM MULTIPROCESS BODY / {__name__} / {num}:{name} {psutil.Process().pid}')
    # 向Queue写入数据
    worker_queue.put(f"子进程 {num} 执行完成")
    time.sleep(1)

def call_multi_process():
    # 创建基于spawn上下文的共享Queue
    ctx = get_context('spawn')
    q = ctx.Queue()
    # 初始化进程池，传递Queue给子进程
    with ProcessPoolExecutor(initializer=init_worker, initargs=(q,)) as pool:
        futures = [pool.submit(run_entry, i) for i in range(3)]
        # 等待所有任务完成
        for future in futures:
            future.result()
        # 从Queue读取子进程返回的数据
        while not q.empty():
            print(q.get())

3. subprocess中实现类似Queue的通信 #

subprocess启动的是独立进程，无法直接使用multiprocessing.Queue，但可以通过管道实现进程间通信，这是最常用的方式：

示例代码

创建worker.py作为子进程入口：

import sys
import time

def worker():
    """从标准输入接收任务，处理后输出到标准输出"""
    for line in sys.stdin:
        num = int(line.strip())
        print(f"Worker 收到任务 {num}")
        time.sleep(1)
        sys.stdout.write(f"任务 {num} 处理完成\n")
        sys.stdout.flush()  # 刷新缓冲区，确保父进程能及时收到数据

if __name__ == '__main__':
    worker()

修改main.py调用subprocess：

import subprocess

def main():
    # 启动子进程，绑定标准输入输出管道
    proc = subprocess.Popen(
        ['python', 'worker.py'],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        text=True
    )
    # 发送任务到子进程
    for i in range(3):
        proc.stdin.write(f"{i}\n")
        proc.stdin.flush()
    proc.stdin.close()  # 关闭输入，告知子进程任务发送完毕
    # 读取子进程输出
    for line in proc.stdout:
        print(line.strip())
    proc.wait()  # 等待子进程结束

if __name__ == '__main__':
    print('i am the GENESIS OF MAIN')
    main()
    print('i am the ENDGAME OF MAIN')

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