跳到主要内容

多进程

多进程是指一个程序同时运行多个独立进程。每个进程都有自己的 Python 解释器和内存空间，因此能够绕开 CPython 的全局解释器锁（GIL），让多个 CPU 核心并行执行计算密集型任务。

不过，多进程并不是“把循环并发执行”这么简单：创建进程、传递数据和回收进程都有成本，进程之间也不能直接读写普通变量。正确使用多进程的关键，是选择合适的任务、减少通信，并明确管理每个子进程的生命周期。

本文以 CPython 3.10 及以上版本为基础。不同 Python 版本和操作系统的默认启动方式可能不同，因此跨平台程序不应依赖某个默认值。

学完后，你应该能够：

  • 判断一个任务是否适合多进程；
  • 使用 Process 创建、等待和停止子进程；
  • 使用 ProcessPoolExecutor 或 Pool 批量分发任务；
  • 使用队列、管道、共享内存和同步原语交换数据；
  • 正确传播异常、设置超时并清理进程资源；
  • 避开序列化、入口保护、死锁和任务粒度等常见问题。

进程、线程与并发​

先区分三个容易混淆的概念：

概念含义
并发多个任务在同一时间段内推进，不要求同一时刻真正一起执行
并行多个任务在同一时刻分别占用计算资源执行
进程操作系统分配资源和隔离程序的一种执行单元，拥有独立地址空间

一个 Python 程序启动后，至少有一个主进程。主进程可以创建子进程，每个子进程都拥有独立的解释器：

主进程
├── 子进程 A：独立解释器、独立内存
├── 子进程 B：独立解释器、独立内存
└── 子进程 C：独立解释器、独立内存

线程通常共享所属进程的内存，交换数据方便，但 CPython 中执行普通 Python 字节码时会受到 GIL 限制。多个进程不共享同一个 GIL，因此更适合纯 Python 的 CPU 密集型计算。

什么时候使用多进程​

适合多进程的典型任务：

  • 图片或音视频转码；
  • 大量独立数据的解析、压缩和计算；
  • 数值模拟、搜索、哈希等 CPU 密集型工作；
  • 需要进程级故障隔离的任务。

不一定适合多进程的任务：

  • 网络请求、数据库等待等 I/O 密集型任务，通常先考虑线程或异步 I/O；
  • 单次执行极快的小任务，调度和序列化成本可能超过计算本身；
  • 需要频繁共享、修改大量内存的任务；
  • 底层库已经自行使用多线程的计算，再开多个进程可能造成 CPU 过度竞争。

判断时不要只看任务数量，要看每个任务的计算量、数据传输量和可独立程度。

使用 Process 创建子进程​

multiprocessing.Process 是最基础的进程 API。下面的程序创建两个子进程并等待它们结束：

from multiprocessing import Process
import os


def calculate_sum(start, end):
    result = sum(range(start, end))
    print(f"进程 {os.getpid()}：{result}")


def main():
    processes = [
        Process(target=calculate_sum, args=(0, 1_000_000)),
        Process(target=calculate_sum, args=(1_000_000, 2_000_000)),
    ]

    for process in processes:
        process.start()

    for process in processes:
        process.join()


if __name__ == "__main__":
    main()

核心步骤只有三个：

  1. 用 Process(target=..., args=...) 描述要执行的函数；
  2. 调用 start() 创建并启动子进程；
  3. 调用 join() 等待子进程结束。

构造 Process 并不会执行目标函数。目标函数是在调用 start() 后，由子进程执行的。

为什么必须保护程序入口​

多进程代码通常必须放在以下保护中：

if __name__ == "__main__":
    main()

使用 spawn 或 forkserver 启动时，子进程会导入主模块。如果创建进程的代码位于模块顶层，导入就会再次创建子进程，最终导致递归启动或运行时错误。

因此，建议始终遵守以下规则：

  • 把创建进程的代码放进函数；
  • 用 if __name__ == "__main__" 保护入口；
  • 把子进程目标函数定义在模块顶层，不要依赖局部函数或 lambda；
  • 在 Windows 可执行程序中按需调用 multiprocessing.freeze_support()。
from multiprocessing import freeze_support


if __name__ == "__main__":
    freeze_support()
    main()

直接运行普通 .py 文件时，freeze_support() 通常不是必需的；它主要用于 PyInstaller 等工具生成的可执行程序。

参数、名称和进程编号​

位置参数通过 args 元组传入，关键字参数通过 kwargs 字典传入：

from multiprocessing import Process
import os


def greet(name, *, prefix):
    print(f"{prefix} {name}，PID={os.getpid()}")


if __name__ == "__main__":
    process = Process(
        target=greet,
        args=("小明",),
        kwargs={"prefix": "你好"},
        name="greeter",
    )
    process.start()
    print(process.name, process.pid)
    process.join()

os.getpid() 返回当前进程的 PID，os.getppid() 返回父进程的 PID。PID 由操作系统管理，只适合诊断，不应当作长期稳定的业务标识。

查看退出状态​

进程结束后，可以检查 exitcode：

process.start()
process.join()

if process.exitcode == 0:
    print("子进程正常结束")
else:
    print(f"子进程异常结束，退出码：{process.exitcode}")
  • None：进程尚未结束；
  • 0：正常结束；
  • 正整数：通常表示程序以非零状态退出；
  • 负整数：在 POSIX 系统上通常表示被对应信号终止。

Process 不会自动把目标函数的返回值或异常对象传回父进程。需要结果和异常传播时，进程池通常更方便。

理解进程的内存隔离​

普通 Python 对象在进程之间不共享。子进程修改自己的变量，不会改变父进程中的对象：

from multiprocessing import Process


numbers = []


def append_number():
    numbers.append(1)
    print("子进程：", numbers)


if __name__ == "__main__":
    process = Process(target=append_number)
    process.start()
    process.join()
    print("父进程：", numbers)  # 通常仍然是 []

有些系统使用 fork 时，子进程在刚启动的一刻看起来拥有父进程数据，但那是操作系统的写时复制机制。双方后续修改的仍是各自的内存，不能把它当作共享变量。

这种隔离带来两面性：

  • 优点：一个进程意外修改内存时，不会直接破坏另一个进程的普通对象；
  • 代价：交换数据必须使用进程间通信，并承担序列化、复制或同步成本。

进程启动方式​

multiprocessing 支持三种主要启动方式：

启动方式工作方式主要特点
spawn启动全新的解释器，再导入主模块隔离清晰、跨平台，但启动较慢，参数必须可序列化
fork复制当前进程启动快，但复制多线程进程可能不安全，仅部分 POSIX 系统支持
forkserver由单线程服务进程负责 fork 子进程减少直接 fork 多线程父进程的风险，仅部分 POSIX 系统支持

支持范围和默认值会随操作系统及 Python 版本变化。可以检查当前环境：

import multiprocessing as mp

print(mp.get_all_start_methods())
print(mp.get_start_method())

应用程序可以显式选择上下文，并用该上下文创建进程、队列和锁：

import multiprocessing as mp


def work(value):
    print(value * value)


if __name__ == "__main__":
    context = mp.get_context("spawn")
    process = context.Process(target=work, args=(5,))
    process.start()
    process.join()

库代码通常不应擅自全局调用 set_start_method()。更好的方式是接受调用者传入的 multiprocessing context，避免与应用程序选择冲突。不同上下文创建的同步对象也不一定能够混用。

使用进程池批量执行任务​

手动创建 Process 适合少量、生命周期明确的进程。需要处理许多同类任务时，应复用固定数量的工作进程，避免为每个任务重复启动解释器。

优先认识 ProcessPoolExecutor​

concurrent.futures.ProcessPoolExecutor 提供了统一、易于处理异常的 Future 接口：

from concurrent.futures import ProcessPoolExecutor


def square(number):
    return number * number


def main():
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = executor.map(square, range(10))
        print(list(results))


if __name__ == "__main__":
    main()

with 代码块结束时会等待任务完成并释放工作进程。max_workers 不宜盲目设置得很大；CPU 密集型任务通常从可用 CPU 数量附近开始测试。

map()：保留输入顺序​

executor.map() 适合将同一个函数应用到多个输入。返回结果的顺序与输入顺序一致，即使后提交的任务先完成：

with ProcessPoolExecutor() as executor:
    results = executor.map(square, [3, 1, 2])
    print(list(results))  # [9, 1, 4]

顺序稳定很方便，但前面的慢任务会推迟后面已完成结果的读取。

submit()：单独管理每个任务​

submit() 返回 Future，适合任务参数不同，或希望按完成顺序处理结果的场景：

from concurrent.futures import ProcessPoolExecutor, as_completed


def divide(left, right):
    return left / right


if __name__ == "__main__":
    jobs = [(10, 2), (8, 4), (6, 3)]

    with ProcessPoolExecutor() as executor:
        future_to_job = {
            executor.submit(divide, left, right): (left, right)
            for left, right in jobs
        }

        for future in as_completed(future_to_job):
            job = future_to_job[future]
            try:
                print(job, future.result())
            except Exception as error:
                print(job, "执行失败：", error)

调用 future.result() 时，子进程中的异常会在父进程中重新抛出，因此应在结果消费位置处理异常。

使用 multiprocessing.Pool​

Pool 提供更传统的多进程池 API：

from multiprocessing import Pool


def square(number):
    return number * number


if __name__ == "__main__":
    with Pool(processes=4) as pool:
        results = pool.map(square, range(10))
    print(results)

常见方法：

方法作用
map()阻塞等待全部任务，结果保持输入顺序
starmap()将每个元素拆成多个位置参数调用函数
imap()惰性、按输入顺序返回结果
imap_unordered()惰性、按完成顺序返回结果
apply_async()异步提交一个函数调用，返回 AsyncResult

多参数任务可以使用 starmap()：

from multiprocessing import Pool


def multiply(left, right):
    return left * right


if __name__ == "__main__":
    with Pool() as pool:
        results = pool.starmap(multiply, [(2, 3), (4, 5), (6, 7)])
    print(results)

新代码若只需要提交任务、获取结果和处理异常，通常优先选择 ProcessPoolExecutor；需要 imap_unordered()、初始化器等 Pool 特有接口时再选 Pool。

合理设置任务粒度​

每次向进程池提交任务，都可能发生序列化和进程间传输。把一百万个微小计算分别提交，通常会非常慢。可以把输入分块，让一次任务完成一批计算：

def sum_squares(numbers):
    return sum(number * number for number in numbers)


chunks = [
    range(0, 250_000),
    range(250_000, 500_000),
    range(500_000, 750_000),
    range(750_000, 1_000_000),
]

with ProcessPoolExecutor() as executor:
    total = sum(executor.map(sum_squares, chunks))

块太小会增加调度成本；块太大会降低负载均衡，并使某个慢任务拖延整体完成。最佳大小需要通过实际数据测试。

可序列化要求​

进程之间经常使用 pickle 序列化任务、参数和结果。这带来几项重要限制：

  • 模块顶层定义的普通函数通常可以传递；
  • lambda、嵌套函数和某些动态对象通常不能可靠序列化；
  • 打开的文件、数据库连接、线程锁等资源通常不能直接传递；
  • 大对象即使能序列化，复制成本也可能很高。

下面的写法不适合作为可移植的进程池任务：

def main():
    factor = 2

    def local_function(value):
        return value * factor

    # local_function 是局部函数，使用 spawn 时通常无法序列化

把函数放到模块顶层，并显式传递所需数据：

def multiply(value, factor):
    return value * factor

此外，不要反序列化不可信来源的数据。pickle 在加载恶意内容时可能执行任意代码。

进程间通信​

多个进程需要协作时，优先通过消息传递交换明确的数据，而不是一开始就共享可变状态。

使用 Queue 传递消息​

multiprocessing.Queue 是多生产者、多消费者的进程安全队列：

from multiprocessing import Process, Queue


def producer(queue):
    for number in range(5):
        queue.put(number * number)
    queue.put(None)  # 结束标记


def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print("收到：", item)


if __name__ == "__main__":
    queue = Queue()
    producer_process = Process(target=producer, args=(queue,))
    consumer_process = Process(target=consumer, args=(queue,))

    producer_process.start()
    consumer_process.start()
    producer_process.join()
    consumer_process.join()

    queue.close()
    queue.join_thread()

队列中的对象通常仍要经过序列化。数据放入队列后，接收方得到的是反序列化后的对象，不是发送方内存中同一个对象。

生产者和消费者应提前约定结束协议，例如发送一个不会与正常数据冲突的哨兵值。多个消费者通常需要为每个消费者发送一个结束标记。

不要依赖 empty() 或 qsize() 判断工作是否结束：并发环境中，状态可能在检查后立即变化，而且这些方法在部分平台上的支持和精确性有限。使用哨兵、任务计数或明确的协调信号更可靠。

使用 Pipe 连接两个端点​

Pipe() 返回两个连接对象，适合两个进程之间直接通信：

from multiprocessing import Pipe, Process


def child(connection):
    value = connection.recv()
    connection.send(value * 2)
    connection.close()


if __name__ == "__main__":
    parent_connection, child_connection = Pipe()
    process = Process(target=child, args=(child_connection,))
    process.start()

    child_connection.close()
    parent_connection.send(21)
    print(parent_connection.recv())  # 42

    parent_connection.close()
    process.join()

如果多个进程同时写同一个管道端点，消息可能损坏。复杂的多生产者、多消费者场景通常应使用 Queue。

使用 Manager 共享代理对象​

Manager 会启动一个服务进程，并提供列表、字典等对象的代理：

from multiprocessing import Manager, Process


def record(shared, key, value):
    shared[key] = value


if __name__ == "__main__":
    with Manager() as manager:
        shared = manager.dict()
        processes = [
            Process(target=record, args=(shared, "a", 1)),
            Process(target=record, args=(shared, "b", 2)),
        ]

        for process in processes:
            process.start()
        for process in processes:
            process.join()

        print(dict(shared))

代理对象使用方便，甚至可以跨机器连接管理器，但每次操作都涉及进程间通信，通常比本地对象慢得多。复合操作也不一定原子，例如“先读取再加一”仍可能产生竞态条件。

共享内存​

大块数据如果反复经过队列复制，成本会很高。此时可以让多个进程访问同一块共享内存，但必须自行约束并发读写。

使用 Value 和 Array​

Value 和 Array 适合共享简单的 ctypes 标量和定长数组：

from multiprocessing import Lock, Process, Value


def increment(counter, lock, times):
    for _ in range(times):
        with lock:
            counter.value += 1


if __name__ == "__main__":
    counter = Value("i", 0)
    lock = Lock()
    processes = [
        Process(target=increment, args=(counter, lock, 10_000))
        for _ in range(4)
    ]

    for process in processes:
        process.start()
    for process in processes:
        process.join()

    print(counter.value)  # 40000

counter.value += 1 包含读取、计算、写回多个步骤，因此即使底层对象带有锁，也应使用同一把显式锁保护整个复合操作。

使用 shared_memory​

multiprocessing.shared_memory 适合共享较大的字节缓冲区，并可与支持缓冲区协议的库配合：

from multiprocessing import Process, shared_memory


def double_first_byte(name):
    existing = shared_memory.SharedMemory(name=name)
    try:
        existing.buf[0] *= 2
    finally:
        existing.close()


if __name__ == "__main__":
    memory = shared_memory.SharedMemory(create=True, size=10)
    try:
        memory.buf[0] = 21
        process = Process(target=double_first_byte, args=(memory.name,))
        process.start()
        process.join()
        print(memory.buf[0])  # 42
    finally:
        memory.close()   # 当前进程不再使用这块内存
        memory.unlink()  # 请求操作系统删除共享内存

共享内存不会自动提供业务级同步。多个进程同时写入时，仍需锁或明确划分互不重叠的数据区域。创建者还必须负责 unlink()，否则资源可能在进程退出后继续占用系统空间。

同步多个进程​

如果多个进程会访问同一资源，就可能发生竞态条件。multiprocessing 提供了与线程同步原语相似的工具：

工具典型用途
Lock同一时刻只允许一个进程进入临界区
RLock同一进程可以重复获取的可重入锁
Semaphore限制同时访问某资源的进程数量
Event广播“某个条件已发生”的开关信号
Condition在锁保护下等待某个状态条件
Barrier让固定数量的进程都到达后再继续

使用锁保护临界区​

from multiprocessing import Lock, Process


def write_line(lock, path, text):
    with lock:
        with open(path, "a", encoding="utf-8") as file:
            file.write(text + "\n")


if __name__ == "__main__":
    lock = Lock()
    processes = [
        Process(target=write_line, args=(lock, "result.txt", f"任务 {index}"))
        for index in range(4)
    ]

    for process in processes:
        process.start()
    for process in processes:
        process.join()

锁只对共同遵守它的代码有效。所有访问同一资源的进程必须使用同一把锁。临界区应尽可能短，并使用 with 确保发生异常时也能释放锁。

使用 Event 通知停止​

from multiprocessing import Event, Process
import time


def worker(stop_event):
    while not stop_event.is_set():
        print("处理中")
        stop_event.wait(0.5)
    print("正常退出")


if __name__ == "__main__":
    stop_event = Event()
    process = Process(target=worker, args=(stop_event,))
    process.start()

    time.sleep(2)
    stop_event.set()
    process.join()

wait(timeout) 同时承担等待和检查信号的作用，比无条件 sleep() 更容易及时响应停止请求。

避免死锁​

死锁通常发生在多个进程互相等待资源时。降低风险的方法包括：

  • 所有代码按固定顺序获取多把锁；
  • 缩短持锁时间，不在持锁期间做慢速 I/O；
  • 使用 with 自动释放锁；
  • 在必要时使用带超时的 acquire(timeout=...)；
  • 不要在等待另一个进程结束时，继续占用它需要的锁或队列资源。

超时、取消和异常​

多进程程序不能假设每个任务都会按时成功。父进程必须为失败、卡住和部分完成设计处理路径。

为等待设置超时​

join(timeout) 超时后只是不再等待，不会自动终止子进程：

process.start()
process.join(timeout=5)

if process.is_alive():
    print("任务超时")

Future 也支持超时：

from concurrent.futures import TimeoutError

try:
    result = future.result(timeout=5)
except TimeoutError:
    print("等待结果超时")

超时意味着调用方停止等待，不一定意味着任务已经停止。Future.cancel() 也只能取消尚未开始运行的任务。

优先协作式退出​

最安全的停止方式是发送 Event 或队列哨兵，让子进程自行结束循环、保存状态并释放资源。实在无法正常停止时，才考虑强制终止：

process.terminate()
process.join()
process.close()

terminate() 会直接终止进程，kill() 在支持的平台上使用更强的终止方式。它们可能导致：

  • finally 和退出处理器来不及运行；
  • 文件、锁或外部事务处于不一致状态；
  • 正在使用的队列或管道损坏；
  • 该进程创建的后代进程继续运行。

因此，强制终止应当是超时后的兜底措施，而不是常规控制流程。close() 只表示当前进程不再使用这个 Process 对象，而且只能在子进程已经结束后调用。

不要忽略子进程异常​

直接使用 Process 时，目标函数异常通常只在子进程中打印，父进程应检查 exitcode，或通过队列传回结构化错误。使用进程池时，务必消费 Future.result()、map() 结果或 AsyncResult.get()，否则异常可能迟迟不被注意。

可以让父进程统一记录任务标识、异常类型和堆栈，而不是让多个子进程直接写同一个日志文件。实际项目中常让子进程通过队列发送日志，再由专门的监听者串行输出。

守护进程​

将 daemon 设为 True，表示子进程应随非守护父进程退出：

process = Process(target=background_work, daemon=True)

守护进程不能再创建子进程，并且在父进程退出时可能被直接终止，无法保证完成清理。因此它只适合可随时丢弃的辅助工作，不适合写入关键数据或必须完成的任务。

这里的 multiprocessing 守护进程只是 Python 进程生命周期属性，不等同于操作系统中的后台服务。

一个完整示例：并行统计文件​

下面的程序并行统计多个文本文件的行数和单词数，并在父进程中集中处理成功与失败结果：

from concurrent.futures import ProcessPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class FileStats:
    path: str
    lines: int
    words: int


def analyze_file(path_text):
    path = Path(path_text)
    text = path.read_text(encoding="utf-8")
    return FileStats(
        path=str(path),
        lines=len(text.splitlines()),
        words=len(text.split()),
    )


def main():
    paths = [str(path) for path in Path("data").glob("*.txt")]

    with ProcessPoolExecutor() as executor:
        future_to_path = {
            executor.submit(analyze_file, path): path
            for path in paths
        }

        for future in as_completed(future_to_path):
            path = future_to_path[future]
            try:
                stats = future.result()
            except (OSError, UnicodeError) as error:
                print(f"{path} 处理失败：{error}")
            else:
                print(
                    f"{stats.path}: "
                    f"{stats.lines} 行，{stats.words} 个单词"
                )


if __name__ == "__main__":
    main()

这个例子体现了几个可复用的设计原则：

  • 工作函数定义在模块顶层，参数和返回值都可序列化；
  • 子进程只分析一个独立输入，不共享可变状态；
  • 父进程负责调度、汇总、报错和输出；
  • 使用上下文管理器保证进程池被关闭；
  • 使用输入路径关联 Future，异常发生时能定位具体任务。

读取少量小文件往往是 I/O 密集型，多进程未必更快；这个示例主要用于展示结构。若文件分析包含耗时的解析或计算，多进程的收益会更明显。

性能测试与调优​

多进程是否更快只能通过测量确定。测试时应关注总耗时、CPU 使用率、内存峰值和吞吐量，并使用接近真实规模的数据。

主要成本​

多进程的额外成本包括：

  • 创建解释器和导入模块；
  • 序列化、复制参数与返回值；
  • 进程间通信与上下文切换；
  • 每个进程独立占用内存；
  • 汇总结果和同步等待。

因此，多进程通常要求单个任务“足够重”。如果串行版本耗时主要花在进程调度之外的计算上，才更可能获得明显收益。

工作进程数量​

CPU 密集型任务可以从 os.process_cpu_count()（较新 Python）或 os.cpu_count() 附近开始测试，但它们表示可用或检测到的逻辑 CPU 数量，不是最佳进程数的保证。

以下情况通常需要减少工作进程：

  • 每个进程占用大量内存；
  • 任务还会调用内部多线程的数值库；
  • 同一机器上还有其他重要服务；
  • 任务争用磁盘、网络或同一个外部系统。

先优化数据流​

常见的有效优化不是增加进程数，而是：

  • 一次传任务标识或文件路径，避免反复传大对象；
  • 合并小任务，减少提交次数；
  • 让子进程返回精简结果；
  • 对大块连续数据使用共享内存，并划分独立写入区域；
  • 避免所有进程竞争同一把锁或同一个输出文件。

常见问题​

程序不断创建子进程​

通常是缺少 if __name__ == "__main__" 入口保护，或者导入模块时执行了创建进程的代码。

报错说对象不能 pickle​

检查传给子进程的目标函数、参数和返回值。优先使用模块顶层函数和普通数据对象，不要传 lambda、局部函数、打开的文件或连接对象。

Jupyter 或交互式环境中无法运行​

交互式环境中的函数不一定能被新解释器正常导入。把工作函数和入口写入 .py 模块后从命令行运行，通常更稳定。

调用了 join() 却一直卡住​

可能原因包括：

  • 子进程在等待永远不会到来的队列消息；
  • 多个进程以不同顺序获取锁，产生死锁；
  • 子进程产生大量队列数据，而父进程先 join()、迟迟不消费；
  • 子进程自身陷入无限循环或阻塞 I/O。

调试时给等待操作增加超时，记录 PID 和任务标识，并检查每个进程当前在等待谁。

多进程反而更慢​

常见原因是任务太小、传输数据太大、进程数过多，或任务本质上受 I/O 和共享资源限制。先测串行基线，再调整进程数和任务分块。

子进程输出混乱​

多个进程同时写标准输出或同一文件时，顺序无法保证。让子进程返回数据，由父进程统一输出；复杂日志则使用进程安全队列集中处理。

测试与调试建议​

  • 先把工作函数写成可单独测试的纯函数；
  • 在单进程中验证正确性，再启用进程池；
  • 测试空输入、单个任务、任务异常和超时；
  • 不要断言并发任务的打印或完成顺序，除非 API 明确保证；
  • 在持续集成中至少覆盖一种使用 spawn 的执行方式，尽早发现序列化和入口问题；
  • 为等待、队列读取和外部 I/O 设置合理超时；
  • 用 PID、任务 ID 和时间戳关联跨进程日志。

多进程测试失败时，最有价值的问题通常不是“哪个进程挂了”，而是“父进程和每个子进程此刻分别在等待什么”。

选择 API 的实用建议​

需求建议选择
少量、长期运行且生命周期不同的进程multiprocessing.Process
批量执行独立函数并收集结果ProcessPoolExecutor
需要 imap_unordered() 等传统池接口multiprocessing.Pool
多生产者、多消费者传递消息multiprocessing.Queue
两个端点直接通信multiprocessing.Pipe
共享少量简单数值Value、Array 加同步原语
共享大块字节数据shared_memory 加明确的所有权与同步规则
共享灵活的字典、列表对象，且性能要求不高Manager 代理对象

可以把设计顺序概括为：

  1. 先确认任务确实是 CPU 密集型且能够拆成独立单元；
  2. 优先使用进程池，而不是无限制创建进程；
  3. 优先传递消息，尽量避免共享可变状态；
  4. 明确处理返回值、异常、超时和关闭流程；
  5. 最后通过基准测试决定进程数和任务粒度。

多进程真正擅长的不是“让任何代码自动变快”，而是把彼此独立、计算量足够大的工作分配给多个 CPU 核心，同时保持清晰的数据边界和生命周期管理。