跳到主要内容

多进程

问题​

Python 多进程怎么用？进程间通信有哪些方式？什么时候用多进程？

答案​

创建进程​

from multiprocessing import Process
import os

def worker(name: str):
    print(f"{name}: PID={os.getpid()}, 父PID={os.getppid()}")

if __name__ == "__main__":  # Windows 下必须有这行
    p = Process(target=worker, args=("子进程",))
    p.start()
    p.join()

进程池​

from multiprocessing import Pool

def cpu_task(n: int) -> int:
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    with Pool(4) as pool:
        # map：保序
        results = pool.map(cpu_task, [1_000_000] * 8)
        
        # apply_async：异步提交
        future = pool.apply_async(cpu_task, (1_000_000,))
        result = future.get(timeout=10)

进程间通信​

Queue：

from multiprocessing import Process, Queue

def producer(q: Queue):
    for i in range(5):
        q.put(i)
    q.put(None)  # 哨兵值

def consumer(q: Queue):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费: {item}")

if __name__ == "__main__":
    q = Queue()
    Process(target=producer, args=(q,)).start()
    Process(target=consumer, args=(q,)).start()

共享内存（3.8+）：

from multiprocessing import shared_memory
import numpy as np

# 创建共享内存
shm = shared_memory.SharedMemory(create=True, size=1000)
arr = np.ndarray((10,), dtype=np.int64, buffer=shm.buf)
arr[:] = range(10)

# 另一个进程可以通过名称访问
shm2 = shared_memory.SharedMemory(name=shm.name)
arr2 = np.ndarray((10,), dtype=np.int64, buffer=shm2.buf)
print(arr2)  # [0, 1, 2, ..., 9]

shm.close()
shm.unlink()

常见面试问题​

Q1: 多线程和多进程怎么选？​

答案：

场景选择原因
CPU 密集型计算多进程绕过 GIL，利用多核
IO 密集型多线程 / asyncioGIL 在 IO 时释放
需要共享大量数据多线程进程间通信开销大
需要隔离性多进程一个进程崩溃不影响其他

Q2: fork 和 spawn 启动方式的区别？​

答案：

启动方式原理平台速度
fork复制父进程（COW）Linux/macOS 默认快
spawn启动新 Python 解释器Windows 默认慢但安全
forkserver用一个服务进程 forkLinux中等

macOS 上推荐 spawn（fork 与某些库不兼容）。

相关链接​