Python Academy · 课时

multiprocessing 基础

Q: 「multiprocessing 基础」这节课中我会学到什么？

使用多个 CPU 核心 你通过在浏览器中直接运行的动手代码来练习 Python Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

使用多个 CPU 核心

multiprocessing 基础 是 CoddyKit 上的免费 Python Academy 课时。这是第 3 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Python Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Python Academy 课程共包含 4 节课。

为什么使用多进程？

GIL 会阻止线程并行运行 Python 字节码。multiprocessing 模块通过启动独立进程来绕过这一限制；每个进程都有自己的解释器和 GIL，因此 CPU 密集型工作可以真正运行在多个核心上。

import multiprocessing as mp
print('CPU cores available:', mp.cpu_count())

启动 Process

Process(target=func) 与线程 API 类似。调用 start() 启动进程，调用 join() 等待进程完成。

import multiprocessing as mp

def worker():
    print('hello from a child process')

if __name__ == '__main__':
    p = mp.Process(target=worker)
    p.start()
    p.join()
    print('parent done')

main 保护机制

在某些平台上，子进程会导入您的脚本。如果没有 if __name__ == '__main__': 保护机制，程序可能会递归地创建进程。请始终将启动代码放在其中。

import multiprocessing as mp

def task(n):
    print('processing', n)

if __name__ == '__main__':
    for i in range(3):
        p = mp.Process(target=task, args=(i,))
        p.start()
        p.join()

传递参数

使用 args 传递位置参数，使用 kwargs 传递关键字参数，方式与线程相同。参数会经过序列化后发送给子进程。

import multiprocessing as mp

def show(label, times):
    for _ in range(times):
        print(label)

if __name__ == '__main__':
    p = mp.Process(target=show, args=('hi',), kwargs={'times': 2})
    p.start()
    p.join()

Pool 工作进程池

multiprocessing.Pool 会管理一组进程。pool.map 会将可迭代对象分配给这些进程，并按顺序收集结果。

import multiprocessing as mp

def square(n):
    return n * n

if __name__ == '__main__':
    with mp.Pool(processes=2) as pool:
        print(pool.map(square, range(6)))

ProcessPoolExecutor

concurrent.futures API 还提供了 ProcessPoolExecutor，其 submit/map 接口与您在线程部分学到的接口相同。

from concurrent.futures import ProcessPoolExecutor

def cube(n):
    return n ** 3

if __name__ == '__main__':
    with ProcessPoolExecutor() as ex:
        print(list(ex.map(cube, range(5))))

独立内存空间

每个进程都有自己的内存。子进程中对全局变量的修改不会改变父进程中的副本。数据必须显式返回或共享。

import multiprocessing as mp

counter = 0

def bump():
    global counter
    counter += 100

if __name__ == '__main__':
    p = mp.Process(target=bump)
    p.start()
    p.join()
    print('parent counter still:', counter)

获取结果

由于内存不共享，请通过 Pool.map、执行器或队列返回结果，而不要修改全局变量。

import multiprocessing as mp

def double(n):
    return n * 2

if __name__ == '__main__':
    with mp.Pool(2) as pool:
        results = pool.map(double, [1, 2, 3, 4])
    print('results:', results)

apply_async 用于单次调用

apply_async 会安排一次调用并返回结果句柄。调用 .get() 可获取其值。

import multiprocessing as mp

def add(a, b):
    return a + b

if __name__ == '__main__':
    with mp.Pool(2) as pool:
        r = pool.apply_async(add, (3, 4))
        print(r.get())

序列化要求

目标函数和参数必须可序列化，这样才能跨越进程边界。顶层函数可以正常工作；lambda 表达式和局部函数通常不行。

import multiprocessing as mp

def top_level(n):
    return n + 1

if __name__ == '__main__':
    with mp.Pool(2) as pool:
        print(pool.map(top_level, [10, 20]))

何时使用进程

对于CPU 密集型工作，请使用进程，例如数学计算、图像处理和解析。创建进程及序列化数据都有开销，因此进程更适合数量较少但负载较重的任务，而不是大量细小的任务。

import multiprocessing as mp

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == '__main__':
    with mp.Pool() as pool:
        print(pool.map(heavy, [1000, 2000, 3000]))

快速检查

请测试您对多进程的理解。

总结

您已学会多进程基础：

进程可以绕过 GIL，实现真正的 CPU 并行。
请使用 if __name__ == '__main__' 保护机制。
Pool 和 ProcessPoolExecutor 会分配工作并收集结果。
内存不共享；参数必须可序列化。

下一步：使用队列和锁安全地共享数据。

常见问题解答