Python 基础体系 · 第 56/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python 多进程：启动方式、IPC、共享内存、Pool 与回收

Python 多进程是指在同一台机器上创建多个操作系统进程，让它们分别运行 Python 代码。与线程不同，每个进程拥有独立的地址空间和 Python 解释器状态，因此进程之间不能直接读取对方的普通变量；它们必须通过进程间通信（Inter-Process Communication，IPC）或显式共享内存交换数据。

multiprocessing 的目标之一，是通过多个子进程绕开传统 CPython 构建中的全局解释器锁（GIL），从而利用多个处理器执行 CPU 密集型 Python 代码。Python 3.14 仍然支持常规 GIL 构建；是否使用多进程，必须结合任务类型、数据传输成本、启动方式和资源生命周期判断，而不能简单理解为“进程一定比线程快”。(docs.python.org)

本文以 Python 3.14 为范围，重点解释以下完整链路：

创建进程
  ↓
选择启动方式：spawn / fork / forkserver
  ↓
准备可导入的目标函数和可序列化参数
  ↓
通过 Queue / Pipe / Manager / SharedMemory 交换数据
  ↓
使用 Process 或 Pool 执行任务
  ↓
处理结果、异常、超时和取消
  ↓
关闭通信对象、等待进程、释放共享资源

一、为什么多进程能够绕开 GIL

1. 普通 Python 线程受什么限制

在传统 CPython 构建中，GIL 是解释器内部用于保护 Python 对象和解释器状态的互斥边界。一个线程执行 Python 字节码时，通常需要持有 GIL；因此多个线程可以并发推进，但纯 Python CPU 计算不一定能在多个 CPU 核心上同时执行。

例如：

def count_down(n: int) -> int:
    while n > 0:
        n -= 1
    return n

如果多个线程同时执行这类纯 Python 循环，它们仍然需要竞争同一个解释器级锁。线程适合 I/O 等待、调用会释放 GIL 的 C 扩展，或者共享同一进程中的对象；但它不是纯 Python CPU 并行的通用方案。

2. 进程的隔离边界

进程的地址空间彼此隔离：

进程 A 地址空间                    进程 B 地址空间
┌──────────────────┐              ┌──────────────────┐
│ Python 对象       │              │ Python 对象       │
│ 堆、栈、解释器状态 │              │ 堆、栈、解释器状态 │
└──────────────────┘              └──────────────────┘
          │                                  │
          └──── Queue / Pipe / SharedMemory ─┘

进程 A 中的：

x = {"count": 1}

不会因为进程 B 也定义了一个名为 x 的变量，就与 B 共享同一个字典。两个字典只是内容可能相同，身份和存储位置完全不同。

因此，进程之间没有一个共享的 Python 对象锁来限制所有解释器。每个进程可以在自己的解释器中执行 Python 字节码，这就是多进程可以实现 CPU 并行的根本原因。

3. 多进程并不消除所有开销

进程并行的实际收益可以粗略表示为：

T_{\text{total}} = T_{\text{startup}} + T_{\text{serialize}} + T_{\text{schedule}} + T_{\text{compute}} + T_{\text{result}}

其中：

$T_{\text{startup}}$ ：创建或启动工作进程的成本；
$T_{\text{serialize}}$ ：参数和结果序列化、反序列化的成本；
$T_{\text{schedule}}$ ：任务排队和调度成本；
$T_{\text{compute}}$ ：真正的计算成本；
$T_{\text{result}}$ ：收集结果的成本。

只有当并行计算节省的时间大于上述额外成本时，多进程才可能带来收益。

例如，把一百万个很小的整数任务分别提交给进程池，可能比单进程循环更慢，因为每个任务的计算时间太短，进程间通信反而成为主要成本。通常应将多个小任务组合成较大的批次，减少任务边界数量。

二、`Process` 的基本模型：创建、启动、等待

multiprocessing.Process 表示一个独立进程。创建 Process 对象本身并不会立即执行目标函数，必须调用 start()；父进程再通过 join() 等待子进程结束。start() 对同一个 Process 对象最多调用一次。(docs.python.org)

# process_basic.py
from multiprocessing import Process
import os


def worker(name: str) -> None:
    print(
        f"name={name}, "
        f"pid={os.getpid()}, "
        f"parent_pid={os.getppid()}"
    )


def main() -> None:
    process = Process(
        target=worker,
        args=("worker-1",),
        name="worker-1",
    )

    process.start()
    process.join()

    print("child pid:", process.pid)
    print("exit code:", process.exitcode)


if __name__ == "__main__":
    main()

运行结果类似：

name=worker-1, pid=12345, parent_pid=12344
child pid: 12345
exit code: 0

这里有三个重要状态：

创建后、启动前：process.pid 为 None，子进程尚未存在；
start() 返回后：操作系统已经创建或安排创建子进程，is_alive() 通常为 True；
子进程结束后：exitcode 不再是 None，正常返回通常为 0。

如果目标函数抛出未捕获异常，子进程通常以非零退出码结束；如果被 POSIX 信号终止，exitcode 可能是对应信号编号的负值。(docs.python.org)

`if name == "main"` 为什么必须存在

使用 spawn 或 forkserver 时，子进程需要导入主模块，以获得目标函数和相关定义。如果模块顶层直接创建新的进程，子进程重新导入该模块时又会创建进程，可能形成递归创建。

错误示例：

# bad.py
from multiprocessing import Process


def worker():
    print("running")


p = Process(target=worker)
p.start()
p.join()

正确写法是把程序入口放进保护块：

from multiprocessing import Process


def worker():
    print("running")


def main():
    p = Process(target=worker)
    p.start()
    p.join()


if __name__ == "__main__":
    main()

这个保护块的含义不是“多进程专用语法”，而是保证：模块被作为主程序运行时才启动任务；模块被子进程导入时，只加载函数和类定义，不重复执行主流程。

目标函数、参数和关键对象通常还需要能够被 pickle。特别是在 spawn 和 forkserver 下，交互式解释器中临时定义的函数、局部函数、闭包和某些不可序列化对象经常导致：

AttributeError: Can't get attribute ...

Python 3.14 不再把 fork 作为任何平台的默认启动方式，因此过去“在 Linux 上直接运行也没问题”的代码，现在更容易暴露出导入和序列化问题。(docs.python.org)

三、三种启动方式：`spawn`、`fork` 与 `forkserver`

启动方式决定子进程如何获得 Python 解释器、模块状态、文件描述符和其他进程资源。Python 3.14 中，启动方式的默认值发生了重要变化：POSIX 平台默认改为 forkserver；Windows 和 macOS 默认仍为 spawn；如果确实需要 fork，必须显式指定。(docs.python.org)

1. `spawn`

spawn 会启动一个全新的 Python 解释器。子进程只继承运行目标所需的资源，通常不会继承父进程中不必要的文件描述符和句柄。

import multiprocessing as mp


def worker(value: int) -> None:
    print(value * 2)


def main() -> None:
    ctx = mp.get_context("spawn")
    process = ctx.Process(target=worker, args=(21,))
    process.start()
    process.join()


if __name__ == "__main__":
    main()

spawn 的特点：

初始化边界清晰；
对导入、序列化和入口保护要求严格；
启动成本通常高于 fork；
是 Windows 和 macOS 的默认方式；
适合需要避免继承父进程复杂状态的程序。

2. `fork`

fork 使用 POSIX 的 os.fork() 创建子进程。子进程开始时几乎拥有父进程的副本，包括内存映射和许多已打开资源。

但“内存副本”不等于“共享同一份可写对象”。操作系统通常使用写时复制（Copy-on-Write）：

fork 后，父子进程暂时指向相同的物理内存页；
任一进程写入某页时，操作系统复制该页；
写入方修改自己的副本；
另一方仍然看到原内容。

所以以下代码不会让 value 在父进程中自动变成 2：

import multiprocessing as mp


def worker(value: int) -> None:
    value += 1
    print("child:", value)


def main() -> None:
    ctx = mp.get_context("fork")
    value = 1

    process = ctx.Process(target=worker, args=(value,))
    process.start()
    process.join()

    print("parent:", value)


if __name__ == "__main__":
    main()

输出类似：

child: 2
parent: 1

fork 的风险在于：如果父进程中已经有线程、锁、数据库连接、网络库内部线程或其他复杂运行时状态，复制进程可能复制出“锁已经被某线程持有，但子进程中该线程不存在”的状态。Python 文档明确指出，安全地 fork 一个多线程进程是有问题的；Python 3.12 起，检测到多线程时相关 fork 操作可能产生 DeprecationWarning。(docs.python.org)

因此，fork 的“启动快”和“继承内存”不能直接等价为“更安全”或“更高效”。

3. `forkserver`

forkserver 会先创建一个专用服务器进程。之后，父进程向该服务器请求创建工作进程，由服务器执行 fork。

由于服务器通常保持单线程，并且不会继承父进程中不必要的资源，forkserver 试图兼顾：

比 spawn 更低的启动成本；
比直接 fork 更少的多线程继承问题；
不继承父进程中无关的文件描述符。

Python 3.14 在支持该方式的 POSIX 平台上将 forkserver 设为默认启动方式。(docs.python.org)

4. 使用上下文，而不是随意修改全局方式

可以用 get_context() 获得一个独立上下文：

import multiprocessing as mp


def worker():
    print("worker")


def main():
    ctx = mp.get_context("forkserver")
    p = ctx.Process(target=worker)
    p.start()
    p.join()


if __name__ == "__main__":
    main()

也可以调用：

mp.set_start_method("spawn")

但全局启动方式通常只能设置一次，且应在主入口中尽早设置。库代码不应擅自改变应用程序的全局启动方式；更适合让调用方传入 mp_context 或上下文对象。不同上下文创建的锁、队列等对象未必可以混用。(docs.python.org)

四、IPC 的核心：消息传递，而不是共享普通变量

进程间通信（IPC）是不同进程交换数据和协调状态的机制。multiprocessing 中常用的 IPC 有：

Queue：多生产者、多消费者队列；
SimpleQueue：简化队列；
JoinableQueue：支持任务完成计数；
Pipe：两个连接端之间的通信；
Connection：更底层的连接对象；
Manager：通过服务器进程和代理对象共享 Python 对象。

这些方式大多需要序列化 Python 对象。队列中放入对象时，对象会被序列化；另一端取出时再反序列化。(docs.python.org)

1. `Queue`：多个生产者和消费者

# queue_ipc.py
from multiprocessing import Process, Queue
from queue import Empty


def producer(queue: Queue) -> None:
    for value in range(5):
        queue.put(("result", value * value))
    queue.put(("done", None))


def consumer(queue: Queue) -> None:
    while True:
        kind, value = queue.get()
        if kind == "done":
            break
        print("received:", value)


def main() -> None:
    queue = Queue()

    p1 = Process(target=producer, args=(queue,))
    p2 = Process(target=consumer, args=(queue,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

    queue.close()
    queue.join_thread()


if __name__ == "__main__":
    main()

数据流为：

producer
   │ queue.put()
   ▼
Queue 的进程安全缓冲区
   │ 序列化后写入管道
   ▼
consumer
   │ queue.get()
   ▼
反序列化后的新对象

输出顺序可能是：

received: 0
received: 1
received: 4
received: 9
received: 16

队列适合任务分发和结果回传，但有两个常被忽略的边界：

Queue 不是零拷贝机制，大对象会产生序列化和复制成本；
子进程向队列写入数据后，后台 feeder thread 可能仍需把缓冲数据刷新到管道，因此进程退出不一定立即发生。父进程如果在数据尚未消费完时 join()，可能出现等待甚至死锁。(docs.python.org)

Queue.empty() 和 Queue.qsize() 也不应被用作严格同步条件。多个进程之间的状态变化具有竞争窗口：

if not queue.empty():
    item = queue.get()

即使检查时非空，另一个消费者也可能在 get() 前取走数据。更可靠的方式是直接 get(timeout=...)，并处理超时或关闭异常。

2. `Pipe`：两个连接端

Pipe() 默认返回一对双工连接：

from multiprocessing import Process, Pipe


def child(conn):
    try:
        request = conn.recv()
        response = {
            "input": request,
            "output": request * request,
        }
        conn.send(response)
    finally:
        conn.close()


def main():
    parent_conn, child_conn = Pipe()

    p = Process(target=child, args=(child_conn,))
    p.start()

    child_conn.close()
    parent_conn.send(12)
    print(parent_conn.recv())

    parent_conn.close()
    p.join()


if __name__ == "__main__":
    main()

输出：

{'input': 12, 'output': 144}

这里父进程关闭了自己不使用的 child_conn，子进程关闭了自己不使用的 parent_conn。关闭无关端点很重要，否则某个进程仍然持有写端时，另一端可能无法及时观察到 EOF。

Pipe(duplex=False) 可以创建单向管道；此时一个端点只能接收，另一个端点只能发送。Pipe 适合两个进程之间的请求—响应或单向流式传输；如果需要多个生产者和消费者，通常选择 Queue。

连接对象的 send() 会序列化对象，recv() 会反序列化对象。recv() 自动反序列化不可信数据存在安全风险，因此不能把未经认证的外部连接当作安全输入。(docs.python.org)

3. `JoinableQueue`：把“取出”和“完成”分开

JoinableQueue 除了 put()、get()，还提供 task_done() 和 join()：

from multiprocessing import Process, JoinableQueue


def worker(queue: JoinableQueue) -> None:
    while True:
        item = queue.get()
        try:
            if item is None:
                return
            print("processing", item)
        finally:
            queue.task_done()


def main():
    queue = JoinableQueue()

    worker_process = Process(target=worker, args=(queue,))
    worker_process.start()

    for item in range(3):
        queue.put(item)

    queue.put(None)

    queue.join()
    worker_process.join()


if __name__ == "__main__":
    main()

状态变化如下：

put(0)      未完成任务数 = 1
put(1)      未完成任务数 = 2
put(2)      未完成任务数 = 3
get(0)
task_done  未完成任务数 = 2
get(1)
task_done  未完成任务数 = 1
get(2)
task_done  未完成任务数 = 0
queue.join() 返回

queue.join() 等待的是所有任务都调用了 task_done()，而不是简单等待队列当前为空。若某条异常路径漏掉 task_done()，主进程可能永久等待，因此通常应放进 finally。

五、共享状态：`Value`、`Array`、`Manager` 与 `SharedMemory`

消息传递强调“发送副本”；共享状态强调“多个进程访问同一份底层数据”。共享状态更接近共享内存模型，但也因此需要处理同步、可见性、竞态和清理问题。

1. `Value` 和 `Array`

Value 用于共享一个 ctypes 值，Array 用于共享一个 ctypes 数组：

from multiprocessing import Process, Value, Array


def worker(number, array):
    number.value = 3.14
    for index in range(len(array)):
        array[index] *= 2


def main():
    number = Value("d", 0.0)
    array = Array("i", [1, 2, 3, 4])

    p = Process(target=worker, args=(number, array))
    p.start()
    p.join()

    print(number.value)
    print(array[:])


if __name__ == "__main__":
    main()

输出：

3.14
[2, 4, 6, 8]

"d" 表示双精度浮点数，"i" 表示有符号整数。Value 和 Array 提供的是有限类型的共享存储，不是任意 Python 对象的共享引用。官方文档说明，这些共享对象带有进程和线程安全支持，但“单次访问安全”不等于“复合操作原子”。(docs.python.org)

下面的计数仍然可能丢失更新：

from multiprocessing import Process, Value


def increment(counter):
    for _ in range(100_000):
        counter.value += 1


def main():
    counter = Value("i", 0)

    processes = [
        Process(target=increment, args=(counter,))
        for _ in range(4)
    ]

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

    print(counter.value)


if __name__ == "__main__":
    main()

逻辑上的 counter.value += 1 至少包含：

读取旧值
计算旧值 + 1
写回新值

两个进程可能同时读取 10，分别计算出 11，最后都写回 11，于是两次增加只留下了一次结果。

应使用显式锁保护整个读—改—写过程：

from multiprocessing import Process, Value


def increment(counter):
    for _ in range(100_000):
        with counter.get_lock():
            counter.value += 1


def main():
    counter = Value("i", 0)

    processes = [
        Process(target=increment, args=(counter,))
        for _ in range(4)
    ]

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

    print(counter.value)  # 400000


if __name__ == "__main__":
    main()

但锁会把临界区串行化。若大量计算都围绕同一个共享计数器进行，进程虽然并行运行，计数更新仍可能成为瓶颈。更常见的替代方案是每个进程先统计自己的局部结果，最后由父进程归并：

# 每个任务返回局部计数
partial_results = [100_000, 100_000, 100_000, 100_000]
total = sum(partial_results)

这体现了一个重要原则：优先传递不可变结果，最后归并；只有确实需要在线共享状态时才使用锁和共享内存。

2. `Manager`：共享对象的代理

Manager() 会创建一个管理器服务器。其他进程拿到的不是原始列表或字典，而是代理对象；对代理对象的方法调用会通过 IPC 转发给管理器进程。

from multiprocessing import Process, Manager


def worker(shared_dict, shared_list):
    shared_dict["status"] = "done"
    shared_list.append(42)


def main():
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()

        p = Process(
            target=worker,
            args=(shared_dict, shared_list),
        )
        p.start()
        p.join()

        print(dict(shared_dict))
        print(list(shared_list))


if __name__ == "__main__":
    main()

输出：

{'status': 'done'}
[42]

Manager 的优势是可以支持较灵活的 Python 类型，例如代理字典、列表、集合、命名空间和同步对象；管理器还可以被不同机器上的进程访问。代价是每次代理操作都可能涉及序列化、网络或进程间请求，因此通常比直接共享内存慢。(docs.python.org)

以下代码看似是一次原子操作：

shared_dict["count"] += 1

但它实际可能拆成：

从 Manager 读取 count
在当前进程计算 +1
向 Manager 写回 count

两个进程仍可能读取到同一个旧值。因此代理对象提供的是访问路径，不会自动把任意复合业务操作变成事务。需要用 Manager 提供的锁，或者改为局部计算后统一归并。

3. `SharedMemory`：直接访问共享字节区域

multiprocessing.shared_memory.SharedMemory 提供一段命名共享内存。创建者和附加者都通过 memoryview 访问同一块底层字节区域：

from multiprocessing import Process
from multiprocessing.shared_memory import SharedMemory


def worker(shm_name: str) -> None:
    shm = SharedMemory(name=shm_name)
    try:
        for index in range(4):
            shm.buf[index] *= 2
    finally:
        shm.close()


def main():
    shm = SharedMemory(create=True, size=4)
    try:
        shm.buf[:4] = bytes([1, 2, 3, 4])

        p = Process(target=worker, args=(shm.name,))
        p.start()
        p.join()

        print(bytes(shm.buf[:4]))  # b'\x02\x04\x06\x08'
    finally:
        shm.close()
        shm.unlink()


if __name__ == "__main__":
    main()

这里有两个不同操作：

close()：关闭当前进程对共享内存的句柄或文件描述符；
unlink()：删除底层命名共享内存块。

每个进程都应在不再访问时调用自己的 close()；unlink() 对同一共享内存块只应调用一次。Windows 上 unlink() 没有效果，底层资源在所有句柄关闭后由系统处理。(docs.python.org)

SharedMemory 只知道字节，不知道这些字节代表整数、浮点数还是矩阵。应用程序必须额外约定：

共享内存布局：
偏移 0～3：元素数量，4 字节整数
偏移 4～35：8 个 double

如果多个进程同时写入同一片区域，仍然需要锁、信号量、事件或其他协议。共享内存减少了复制，却没有自动解决：

谁可以写；
什么时候写完；
读者如何知道数据已准备好；
多个字段如何保持一致；
写入进程崩溃后数据处于什么状态。

4. `SharedMemoryManager`

如果程序中共享内存对象较多，可以使用 SharedMemoryManager 管理它们的生命周期。它适合把“分配共享内存”和“最终释放共享内存”绑定到一个上下文中，但仍需保证附加进程不在管理器退出后继续访问这些区域。

Python 3.14 的 SharedMemory 还涉及资源跟踪器。由 multiprocessing 创建的相关进程通常共享一个 resource tracker；而通过其他方式启动的独立 Python 进程可能拥有自己的 tracker。此时多个 tracker 可能对同一共享内存的所有权产生冲突。Python 3.13 增加了 track 参数；Windows 则有自己的句柄跟踪机制。(docs.python.org)

六、同步原语：锁不是 IPC 的替代品

multiprocessing 提供与 threading 类似的同步原语：

Lock：互斥锁；
RLock：可重入锁；
Semaphore：信号量；
BoundedSemaphore：有界信号量；
Event：事件标志；
Condition：条件变量；
Barrier：屏障。

同步原语解决的是“谁能在什么时候访问某个状态”，IPC 解决的是“如何把数据或信号传给另一个进程”。两者经常一起出现。

例如，使用 Event 通知工作进程停止：

from multiprocessing import Event, Process
import time


def worker(stop_event: Event):
    while not stop_event.is_set():
        print("working")
        time.sleep(0.2)

    print("stopped")


def main():
    stop_event = Event()
    p = Process(target=worker, args=(stop_event,))
    p.start()

    time.sleep(1)
    stop_event.set()

    p.join()


if __name__ == "__main__":
    main()

这个例子中：

父进程 set()
    ↓
共享事件状态变为 set
    ↓
子进程下一次检查 is_set() 为 True
    ↓
子进程自行退出

这是一种协作式停止。它不会强行打断正在执行的不可中断操作，目标函数必须主动检查事件。

七、Pool：把“创建进程”变成“提交任务”

直接使用 Process 时，应用程序需要自行管理：

创建多少个进程；
给哪个进程分配什么任务；
结果如何返回；
失败如何处理；
什么时候回收。

multiprocessing.Pool 把这些工作封装为工作进程池。父进程提交任务，池中的 worker 进程从任务队列获取输入，执行函数，再把结果放入结果通道。Pool 默认使用 os.process_cpu_count() 作为进程数来源；也可以显式传入 processes。(docs.python.org)

1. `map`：有序、阻塞

from multiprocessing import Pool


def square(value: int) -> int:
    return value * value


def main():
    with Pool(processes=2) as pool:
        result = pool.map(square, range(6))

    print(result)


if __name__ == "__main__":
    main()

输出：

[0, 1, 4, 9, 16, 25]

map() 的特点：

接收一个可迭代对象；
把每个元素作为参数传给函数；
等待所有任务完成；
返回结果顺序与输入顺序一致。

如果任务耗时分别为：

任务 A：5 秒
任务 B：1 秒
任务 C：1 秒

即使 B、C 先完成，map() 仍然按照输入顺序组织结果。调用方在等待 A 时，无法通过 map() 直接消费已经完成的 B、C。

2. `imap`：惰性产生有序结果

from multiprocessing import Pool
import time


def work(value: int) -> int:
    time.sleep(value * 0.1)
    return value


def main():
    with Pool(2) as pool:
        for result in pool.imap(work, [3, 1, 2]):
            print("result:", result)


if __name__ == "__main__":
    main()

imap() 返回迭代器，结果仍保持输入顺序，但不会一次性把所有结果构造成列表。适合输入规模较大、希望逐步消费结果的场景。

3. `imap_unordered`：谁先完成谁先返回

from multiprocessing import Pool
import time


def work(value: int) -> int:
    time.sleep(value * 0.1)
    return value


def main():
    with Pool(2) as pool:
        for result in pool.imap_unordered(work, [3, 1, 2]):
            print("result:", result)


if __name__ == "__main__":
    main()

输出顺序可能是：

result: 1
result: 3
result: 2

实际顺序受 worker 调度影响。它适合结果彼此独立、调用方不要求输入顺序的任务。

4. `apply_async`：提交单个异步任务

from multiprocessing import Pool


def divide(a: int, b: int) -> float:
    return a / b


def main():
    with Pool(2) as pool:
        async_result = pool.apply_async(divide, args=(10, 2))

        print("task submitted")
        print("result:", async_result.get(timeout=3))


if __name__ == "__main__":
    main()

apply_async() 返回 AsyncResult。调用 get() 时才等待结果：

任务成功：返回函数结果；
任务失败：get() 重新抛出远程调用中的异常；
超过 timeout：抛出 multiprocessing.TimeoutError。

from multiprocessing import Pool, TimeoutError
import time


def slow_task():
    time.sleep(10)
    return "done"


def main():
    with Pool(1) as pool:
        result = pool.apply_async(slow_task)

        try:
            print(result.get(timeout=1))
        except TimeoutError:
            print("timeout")


if __name__ == "__main__":
    main()

这里的超时只表示父进程等待结果超时，不等于已经取消或杀死 worker。任务可能仍在后台执行，池仍然可用。(docs.python.org)

5. `starmap`：展开多个参数

from multiprocessing import Pool


def add(a: int, b: int) -> int:
    return a + b


def main():
    pairs = [(1, 2), (3, 4), (5, 6)]

    with Pool(2) as pool:
        result = pool.starmap(add, pairs)

    print(result)  # [3, 7, 11]


if __name__ == "__main__":
    main()

starmap() 等价于把每个元组展开后调用：

add(1, 2)
add(3, 4)
add(5, 6)

6. `chunksize` 的作用

Pool 可能把输入任务按块发送给 worker，而不是逐个发送。chunksize 越大：

任务提交次数减少；
序列化和调度开销可能下降；
结果粒度变粗；
任务耗时不均衡时可能造成某些 worker 长时间忙碌、其他 worker 空闲。

假设有 100 个任务和 4 个 worker：

chunksize=1：
任务边界细，调度灵活，但 IPC 次数多

chunksize=25：
IPC 次数少，但若某一块特别慢，其他 worker 可能提前空闲

因此，chunksize 不是越大越好。均匀的小任务可以适当增大；耗时差异大的任务应保留更细的调度粒度。

八、Pool 的初始化与 worker 回收

1. `initializer`：每个 worker 启动时执行一次

from multiprocessing import Pool
import os

RESOURCE = None


def init_worker(prefix: str) -> None:
    global RESOURCE
    RESOURCE = f"{prefix}-pid-{os.getpid()}"


def task(value: int) -> str:
    return f"{RESOURCE}: {value * value}"


def main():
    with Pool(
        processes=2,
        initializer=init_worker,
        initargs=("worker",),
    ) as pool:
        print(pool.map(task, [1, 2, 3, 4]))


if __name__ == "__main__":
    main()

initializer 适合初始化每个 worker 私有的资源，例如：

打开只读配置；
建立每个进程独立的数据库连接；
加载模型；
设置进程级日志上下文。

不要在父进程创建一个不能安全跨进程继承的连接，然后期待所有 worker 共享它。更稳妥的方式是让每个 worker 在自己的初始化函数中创建自己的资源。

2. `maxtasksperchild`：限制 worker 的任务寿命

Pool worker 默认可以持续存在到整个 Pool 生命周期。maxtasksperchild 指定一个 worker 最多执行多少个任务，达到上限后退出，并由 Pool 创建新的 worker 替换。其目的之一是释放长期累积的资源。(docs.python.org)

from multiprocessing import Pool


def task(value):
    return value * value


def main():
    with Pool(
        processes=2,
        maxtasksperchild=100,
    ) as pool:
        result = pool.map(task, range(1000))

    print(len(result))


if __name__ == "__main__":
    main()

它可以缓解某些长期运行任务中的：

C 扩展内存碎片；
第三方库内部资源累积；
不易定位的进程级泄漏；
单个 worker 状态逐渐污染。

但 worker 重建也有成本：

worker 退出
  ↓
重新启动解释器
  ↓
重新导入模块
  ↓
重新执行 initializer
  ↓
继续领取任务

因此，maxtasksperchild 是资源回收策略，不是默认性能优化。设置得过小会使启动和初始化成本明显增加。

九、进程和 Pool 的回收语义

“任务完成”与“进程资源已释放”不是同一个概念。必须区分：

任务函数返回；
worker 进程退出；
父进程调用 join()；
队列后台线程完成刷新；
共享内存调用 unlink()；
Pool 或 Executor 关闭。

1. 单个 `Process`：`join()` 与 `close()`

典型生命周期是：

创建 Process
   ↓
start()
   ↓
运行
   ↓
join()
   ↓
确认 exitcode
   ↓
close()

from multiprocessing import Process


def worker():
    print("done")


def main():
    p = Process(target=worker)
    p.start()

    p.join()

    if p.exitcode != 0:
        raise RuntimeError(f"child failed: {p.exitcode}")

    p.close()


if __name__ == "__main__":
    main()

close() 释放的是父进程中 Process 对象关联的资源；如果底层子进程仍在运行，调用 close() 会抛出 ValueError。因此通常先 join()，确认进程已经结束，再 close()。(docs.python.org)

2. Pool：`close()`、`terminate()`、`join()`

Pool 有三种关键操作：

`close()`

不再接受新任务。已经提交的任务会继续完成，worker 完成后退出。

pool.close()
pool.join()

`terminate()`

立即停止 worker，不等待未完成任务。正在使用管道、队列、锁或信号量的进程被强行终止时，共享资源可能损坏、锁可能保持占用、队列可能不可用。(docs.python.org)

`join()`

等待 worker 进程退出。调用 join() 前必须先调用 close() 或 terminate()。

手动管理时：

from multiprocessing import Pool


def task(value):
    return value * value


def main():
    pool = Pool(2)
    try:
        result = pool.map(task, range(5))
        print(result)
    except BaseException:
        pool.terminate()
        raise
    else:
        pool.close()
    finally:
        pool.join()


if __name__ == "__main__":
    main()

语义是：

成功路径：
提交任务 → 获取结果 → close → join

失败路径：
发生异常 → terminate → join → 重新抛出异常

3. `with Pool(...)` 的真实含义

with Pool(2) as pool:
    result = pool.map(task, data)

退出上下文时会调用 Pool 的终止逻辑，而不是简单等价于“先 close() 再 join()”。官方文档说明，Pool 的上下文管理器退出时调用 terminate()。因此：

如果代码正常执行完，通常问题不大；
如果上下文中仍有未完成任务，退出时可能直接终止它们；
如果需要保证所有已提交任务完成，应显式设计成功路径，或在离开上下文前等待任务全部取得结果。(docs.python.org)

十、取消、超时和强制终止不是一回事

这三个概念经常被混淆。

1. 超时

result.get(timeout=1)

表示调用方最多等待 1 秒。超时后，任务可能仍在执行。

2. 取消

对于 Pool.AsyncResult，没有与 Future.cancel() 完全相同的通用取消接口。已经进入 worker 执行阶段的任务，通常不能安全地从外部撤销；可以通过协作式停止标志让任务自行结束。

3. 终止

process.terminate()
process.kill()

这是强制停止进程的操作。terminate() 在 POSIX 上使用 SIGTERM，Windows 上使用 TerminateProcess()；kill() 在 POSIX 上使用 SIGKILL。它们不会保证执行 finally、退出处理器或业务清理代码，也可能破坏正在使用的队列、管道和锁。(docs.python.org)

Python 3.14 新增了 Process.interrupt()。在 POSIX 上，它通过 SIGINT 让子进程默认产生 KeyboardInterrupt；如果子进程捕获并忽略该异常，进程不一定退出。在 Windows 上该行为未定义，因此不能把它作为跨平台强制取消机制。(docs.python.org)

十一、`ProcessPoolExecutor` 与 `Pool` 的关系

multiprocessing.Pool 是直接面向进程池的 API；concurrent.futures.ProcessPoolExecutor 是更高层的任务执行器。二者都使用进程池，但抽象不同：

方面	`multiprocessing.Pool`	`ProcessPoolExecutor`
结果对象	`AsyncResult`	`Future`
典型提交方式	`map`、`apply_async`	`submit`、`map`
取消接口	较弱	`Future.cancel()`
与线程池统一性	较低	较高
进程池生命周期	`close/terminate/join`	`shutdown` 或上下文管理器
进程上下文	`context`	`mp_context`

如果应用已经使用 Future、回调、等待和取消模型，ProcessPoolExecutor 往往更容易和 ThreadPoolExecutor 统一；如果需要 Pool 的 initializer、maxtasksperchild、imap_unordered 等行为，直接使用 Pool 更直观。

`ProcessPoolExecutor` 端到端示例

# executor_example.py
from concurrent.futures import (
    ProcessPoolExecutor,
    as_completed,
)
import multiprocessing as mp


def square(value: int) -> int:
    return value * value


def main():
    ctx = mp.get_context("spawn")

    with ProcessPoolExecutor(
        max_workers=2,
        mp_context=ctx,
    ) as executor:
        futures = [
            executor.submit(square, value)
            for value in range(6)
        ]

        for future in as_completed(futures):
            try:
                print("result:", future.result())
            except Exception as exc:
                print("task failed:", repr(exc))


if __name__ == "__main__":
    main()

submit() 返回 Future。Future 的状态可以抽象为：

PENDING
  ├── cancel() 成功 → CANCELLED
  └── worker 开始执行 → RUNNING
                            ├── 正常返回 → FINISHED
                            └── 抛出异常 → FINISHED（带异常）

cancel() 只能取消尚未开始运行的任务；已经运行的任务不能通过 cancel() 强制停止。future.result() 会在任务成功时返回值，在任务失败时重新抛出异常。

`shutdown`

executor.shutdown(
    wait=True,
    cancel_futures=True,
)

含义是：

wait=True：等待已经开始运行的任务完成；
cancel_futures=True：取消尚未开始运行的待处理任务；
已经运行或已经完成的任务不会被 cancel_futures 取消。

使用 with ProcessPoolExecutor(...) 会在退出时执行等待式关闭。(docs.python.org)

Python 3.14 的 worker 批量终止

Python 3.14 为 ProcessPoolExecutor 增加了：

executor.terminate_workers()
executor.kill_workers()

前者尝试对所有存活 worker 调用 Process.terminate()，后者调用 Process.kill()；两者内部还会执行 executor 的关闭逻辑。调用后不能继续提交任务。它们适合进程池进入不可恢复状态或应用需要快速放弃剩余任务的场景，但同样不能保证业务清理代码执行。(docs.python.org)

十二、异常路径：子进程失败后父进程看到了什么

1. 直接使用 `Process`

子进程异常不会自动以父进程异常的形式抛出。父进程必须检查：

p.join()

if p.exitcode != 0:
    raise RuntimeError(
        f"worker failed: pid={p.pid}, exitcode={p.exitcode}"
    )

如果需要传递详细异常信息，应显式把错误消息放入队列或管道：

from multiprocessing import Process, Queue


def worker(queue):
    try:
        result = 10 / 0
    except Exception as exc:
        queue.put({
            "ok": False,
            "error_type": type(exc).__name__,
            "message": str(exc),
        })
    else:
        queue.put({"ok": True, "result": result})


def main():
    queue = Queue()
    p = Process(target=worker, args=(queue,))
    p.start()

    message = queue.get()
    p.join()

    if not message["ok"]:
        raise RuntimeError(
            f'{message["error_type"]}: {message["message"]}'
        )


if __name__ == "__main__":
    main()

2. Pool 和 Future

Pool 的 AsyncResult.get() 会重新抛出远程调用异常；Future.result() 也会把任务异常交给调用方。调用方必须实际读取结果，否则失败可能只停留在异步结果对象中，没有进入主流程的错误处理路径。(docs.python.org)

3. worker 进程突然退出

如果 ProcessPoolExecutor 的 worker 异常退出，executor 会报告 BrokenProcessPool，而不是让调用方无限等待。Python 3.14.7 还修复了 worker 达到 max_tasks_per_child 后、队列仍有任务时可能出现的死锁问题。(docs.python.org)

十三、一个完整的共享内存并行例子

下面使用共享内存存放整数数组，多个进程分别处理不同区间。任务参数只传递共享内存名称、数组长度和区间边界，不复制整个数组。

# shared_array_workers.py
from multiprocessing import Process
from multiprocessing.shared_memory import SharedMemory
import struct


ITEM_SIZE = 8  # 一个 64 位整数占 8 字节


def write_int(buffer, index: int, value: int) -> None:
    offset = index * ITEM_SIZE
    struct.pack_into("q", buffer, offset, value)


def read_int(buffer, index: int) -> int:
    offset = index * ITEM_SIZE
    return struct.unpack_from("q", buffer, offset)[0]


def worker(shm_name: str, start: int, end: int) -> None:
    shm = SharedMemory(name=shm_name)
    try:
        for index in range(start, end):
            value = read_int(shm.buf, index)
            write_int(shm.buf, index, value * value)
    finally:
        shm.close()


def main():
    values = list(range(10))
    shm = SharedMemory(create=True, size=len(values) * ITEM_SIZE)

    try:
        for index, value in enumerate(values):
            write_int(shm.buf, index, value)

        processes = []
        block_size = 5

        for start in range(0, len(values), block_size):
            end = min(start + block_size, len(values))
            p = Process(
                target=worker,
                args=(shm.name, start, end),
            )
            p.start()
            processes.append(p)

        for p in processes:
            p.join()
            if p.exitcode != 0:
                raise RuntimeError(
                    f"worker failed: exitcode={p.exitcode}"
                )
            p.close()

        result = [
            read_int(shm.buf, index)
            for index in range(len(values))
        ]
        print(result)

    finally:
        shm.close()
        shm.unlink()


if __name__ == "__main__":
    main()

预期输出：

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

这个例子能够安全工作，是因为每个进程写入不重叠的区间：

worker 1：索引 0～4
worker 2：索引 5～9

如果两个进程同时写入同一个索引，就需要额外同步；否则即使每次底层字节写入看起来很小，也不能把整个业务更新过程视为自动原子。

十四、常见失败表现与诊断路径

1. 子进程不断递归创建

表现：

子进程数量快速增加
程序反复重新启动
RuntimeError 或系统资源耗尽

通常原因是缺少：

if __name__ == "__main__":

或者在模块导入阶段就创建了进程池。

诊断方式：

查看入口模块是否有保护块；
检查全局变量初始化是否隐式创建 Pool、Queue 或 Process；
用 spawn 显式运行测试，尽早暴露导入问题。

2. `Can't get attribute`

表现：

AttributeError: Can't get attribute 'worker' on module '__main__'

通常原因：

目标函数在交互式解释器中定义；
目标函数是局部函数；
使用 lambda；
模块不可导入；
参数包含不可 pickle 的对象。

处理方式是把 worker 函数放到可导入模块的顶层，并让参数成为可序列化数据。

3. 父进程在 `join()` 中卡住

常见原因包括：

子进程向 Queue 写入大量数据，但父进程没有消费；
父进程先 join()，后 get()；
Queue 的 feeder thread 尚未刷新完；
某个进程持有锁后异常退出；
Pool 未调用 close() 或 terminate()。

对于队列，通常应先确保消费者持续消费，再等待生产者退出：

启动生产者
启动消费者
生产者 put
消费者 get
消费者完成
生产者 join
消费者 join

而不是：

生产者启动
生产者 join
父进程最后才读取 Queue

4. 进程池任务似乎没有并行

可能原因：

每个任务太小，IPC 开销超过计算时间；
chunksize 不合适；
worker 数量超过有效 CPU 并行度；
任务实际在等待锁、磁盘、网络或数据库；
父进程在按输入顺序等待慢任务；
创建进程和加载资源时间占主导。

诊断时应分别测量：

单进程计算时间
进程池启动时间
任务序列化时间
worker 实际计算时间
结果收集时间

只测总墙钟时间，很难知道瓶颈在计算还是通信。

5. 强制终止后后续任务继续报错

如果 worker 在写 Queue、Pipe 或持有锁时被 terminate() 或 kill()，其他进程可能观察到损坏的通信对象或永远无法获得的锁。强制终止应作为不可恢复路径，而不是普通超时处理。(docs.python.org)

十五、如何选择进程、线程、异步和进程池

可以用任务的主要等待对象来判断：

任务特征	通常优先考虑
纯 Python CPU 计算	多进程或 `ProcessPoolExecutor`
阻塞 I/O，且库允许线程安全调用	线程池
大量网络连接、协作式 I/O	`asyncio`
单个外部命令或独立程序	`subprocess`
大型数值计算且底层释放 GIL	先评估线程或库自身并行
大数据在进程间反复传输	批处理、共享内存或专用数据通道
需要统一 Future 接口	`concurrent.futures`
需要细粒度 Pool 行为	`multiprocessing.Pool`

多进程尤其适合这样的函数：

f(x) \rightarrow y

其中每个输入 $x$ 可以独立处理，输出 $y$ 较小，任务之间不需要频繁访问共享可变状态。此时数据流可以组织为：

输入分片
  ↓
独立 worker
  ↓
局部结果
  ↓
父进程归并

如果任务的核心逻辑是：

读取共享状态
修改共享状态
读取另一个共享状态
再次修改

那么进程数增加后，锁竞争、代理调用和一致性协议可能抵消并行收益。此时应重新设计数据流，而不是单纯增加 worker 数量。

十六、Python 3.14 下的关键记忆点

multiprocessing 使用进程隔离来绕开传统 CPython GIL 对纯 Python CPU 并行的限制，但 IPC、序列化和启动本身都有成本。
Python 3.14 中，POSIX 默认启动方式是 forkserver；Windows 和 macOS 默认是 spawn；需要 fork 时必须显式指定。(docs.python.org)
spawn 和 forkserver 要求主模块可导入，入口必须使用 if __name__ == "__main__"。
Queue 和 Pipe 传递的是序列化后的对象，不是共享 Python 对象。
Value、Array 和 SharedMemory 可以共享底层数据，但不自动解决复合操作的竞态。
Manager 通过服务器进程和代理对象提供灵活共享，代价是额外 IPC，通常比直接共享内存慢。(docs.python.org)
Pool.map() 有序且阻塞；imap_unordered() 按完成顺序产生结果；apply_async() 返回异步结果对象。
Pool 的正常收尾是 close() 后 join()；异常或放弃任务时使用 terminate()，但强制终止可能破坏队列、管道和锁。
maxtasksperchild 可以周期性替换 worker，释放长期累积的进程级资源，但会增加重启成本。
SharedMemory.close() 和 unlink() 负责不同层面的资源释放，不能只依赖对象析构或垃圾回收。
ProcessPoolExecutor 使用 Future 表达任务状态，并提供 shutdown(cancel_futures=True)；Python 3.14 还提供了 terminate_workers() 和 kill_workers()。(docs.python.org)
多进程设计的关键不是“启动更多进程”，而是让任务边界足够粗、数据传输足够少、共享状态足够少、失败路径足够明确。

系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。

Python 多进程：启动方式、IPC、共享内存、Pool 与回收

一、为什么多进程能够绕开 GIL

1. 普通 Python 线程受什么限制

2. 进程的隔离边界

3. 多进程并不消除所有开销

二、Process 的基本模型：创建、启动、等待

if __name__ == "__main__" 为什么必须存在

三、三种启动方式：spawn、fork 与 forkserver

1. spawn

2. fork

3. forkserver

4. 使用上下文，而不是随意修改全局方式

四、IPC 的核心：消息传递，而不是共享普通变量

1. Queue：多个生产者和消费者

2. Pipe：两个连接端

3. JoinableQueue：把“取出”和“完成”分开

五、共享状态：Value、Array、Manager 与 SharedMemory

1. Value 和 Array

2. Manager：共享对象的代理

3. SharedMemory：直接访问共享字节区域

4. SharedMemoryManager