Python 进程池启动报错：用 spawn、main 入口保护与异常测试修复

把串行脚本换成进程池后，程序在 Linux 上报 RuntimeError，提示启动阶段又创建了进程；把函数改成 lambda，又出现无法 pickle 的错误。问题常常不在计算本身，而在子进程如何重新导入脚本。

本文面向初级到中级读者，做一个显式使用 spawn 的 CPU 批处理入口：分块统计素数，比较串行和进程池结果，再验证参数错误和工作进程异常。示例不改系统设置，不安装依赖，也不承诺小任务一定加速。

1. 方法：把计算函数和程序入口分开

计算函数放在模块顶层，参数和返回值只用简单数据。
参数解析、创建进程池和输出报告都从 main() 进入，用 if __name__ == "__main__": 保护入口。
显式传入 mp.get_context("spawn")，不依赖当前系统的默认启动方式。
消费全部计算结果之后，主进程才打印成功 JSON。

Python 的安全导入说明解释了入口保护的原因：spawn 会启动新的解释器，子进程导入主模块时不能再次启动进程。不要把工作函数也放进入口保护块，否则导入时没有该函数。

Python 3.14 的 ProcessPoolExecutor 文档记录了默认启动方式不再使用 fork。这里固定 spawn 是为了复现一致的导入规则，不是要求所有应用都改成 spawn。

本文处理纯 Python 计算示例。大量文件读取可先看有界线程池批处理；脚本慢在哪里还不清楚时，先用 cProfile 定位热点，不要先堆进程数。

2. 准备：从真实脚本文件运行

需要现有 Python 3.9 或更新版本，无第三方包。命令在 Linux、CPython 3.11.4 下测试，可在 bash 或 zsh 中执行；没有实测 Windows 或其他 Python 版本。

在同一个终端创建新的练习目录，后续命令都在该目录执行：

python3 --version
mkdir processpool-demo
cd processpool-demo

把下一节完整代码保存为 count_primes.py，或下载 count_primes.py。不要把创建进程池的整段脚本直接粘贴进 REPL、Notebook 或 python3 - <<'PY'；本例要求子进程能导入磁盘上的模块。

3. 实现：显式 spawn、分块与统一汇总

"""Count primes with an import-safe process pool. Python 3.9+, standard library."""
import argparse
from concurrent.futures import ProcessPoolExecutor
import json
from math import isqrt
import multiprocessing as mp
import sys


def count_chunk(bounds):
    start, stop = bounds
    if start < 2 or stop < start:
        raise ValueError("invalid interval")
    count = 0
    for number in range(start, stop):
        if number == 2 or (number % 2 and all(
            number % divisor for divisor in range(3, isqrt(number) + 1, 2)
        )):
            count += 1
    return count


def calculate(chunks, workers, serial=False):
    if serial:
        return sum(map(count_chunk, chunks))
    with ProcessPoolExecutor(
        max_workers=workers, mp_context=mp.get_context("spawn")
    ) as pool:
        return sum(pool.map(count_chunk, chunks))


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--limit", type=int, default=10000)
    parser.add_argument("--chunk-size", type=int, default=2500)
    parser.add_argument("--workers", type=int, default=2)
    parser.add_argument("--serial", action="store_true", help="skip the pool")
    args = parser.parse_args()
    if not 2 <= args.limit <= 1000000:
        parser.error("--limit must be 2..1000000")
    if not 1 <= args.chunk_size <= 1000000:
        parser.error("--chunk-size must be 1..1000000")
    if not 1 <= args.workers <= 8:
        parser.error("--workers must be 1..8")
    task_count = (args.limit - 1 + args.chunk_size - 1) // args.chunk_size
    if task_count > 1000:
        parser.error("too many chunks; increase --chunk-size (maximum 1000 chunks)")
    chunks = [
        (start, min(start + args.chunk_size, args.limit + 1))
        for start in range(2, args.limit + 1, args.chunk_size)
    ]
    try:
        total = calculate(chunks, args.workers, args.serial)
    except (OSError, RuntimeError, ValueError) as exc:
        print(f"ERROR: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1
    print(json.dumps({"limit": args.limit, "chunks": len(chunks),
                      "prime_count": total}))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

count_chunk() 统计半开区间 [start, stop) 内的素数。主进程覆盖从 2 到 --limit 的所有整数，包含上限，没有重叠；默认产生 4 个任务。每个工作进程只返回整数，不写文件、不打印成功标记。

这里的 --chunk-size 是每个计算任务覆盖的整数数量，不是 Executor.map() 的同名 chunksize 参数。最多接受 1000 个任务、8 个工作进程和 100 万的上限，防止练习命令意外生成大量提交；这些是示例的策略限制，不是 Python 的平台限制。

sum(pool.map(...)) 会取出所有结果，工作函数的异常会在取结果时重新抛出。with 负责关闭进程池，但退出时会等待任务结束，不是硬超时或立即停止机制。这些行为见 Executor 的官方说明。

4. 使用：先验证输出，再测时间

查看帮助并分别执行串行和进程池模式：

python3 count_primes.py --help
python3 count_primes.py --serial
python3 count_primes.py --workers 2

后两条命令都应输出同一行，退出码为 0：

{"limit": 10000, "chunks": 4, "prime_count": 1229}

--workers 1 仍然启动一个工作进程，只有 --serial 完全不创建进程池。参数非法时 argparse 返回 2；代码捕获到的资源、运行时或区间错误返回 1，写入 stderr，不打印成功 JSON。其他未捕获异常也会以非零状态和回溯失败，不应视为成功。

需要评估性能时，在 bash 中对同一输入分别运行几次：

time python3 count_primes.py --limit 200000 --chunk-size 25000 --serial
time python3 count_primes.py --limit 200000 --chunk-size 25000 --workers 2

比较实际 real 时间，同时确认两种模式的 JSON 相同。这个素数算法只是可验证的 CPU 练习，不是高性能素数筛；默认小输入常被解释器启动和进程间通信开销淹没。CPU 配额、任务大小和其他负载都会影响结果。

5. 验证：成功、参数错误与工作进程异常

把下面代码保存为 verify_pool.py 文件，或下载 verify_pool.py。验证脚本本身也需要入口保护，不能作为 stdin 脚本运行。

"""Run from a saved .py file, not stdin: spawn must import the main module."""
from concurrent.futures import ProcessPoolExecutor
import json
import multiprocessing as mp
import subprocess
import sys
from count_primes import count_chunk


def main():
    expected = {"limit": 10000, "chunks": 4, "prime_count": 1229}
    for extra in (["--serial"], [], ["--workers", "1"]):
        result = subprocess.run(
            [sys.executable, "count_primes.py", *extra],
            capture_output=True, text=True, timeout=30, check=False,
        )
        assert result.returncode == 0 and not result.stderr, result
        assert json.loads(result.stdout) == expected, result
    for extra in (["--workers", "0"], ["--limit", "1"],
                  ["--chunk-size", "0"], ["--chunk-size", "1"]):
        result = subprocess.run(
            [sys.executable, "count_primes.py", *extra],
            capture_output=True, text=True, timeout=30, check=False,
        )
        assert result.returncode == 2 and result.stdout == "", result
        assert "error:" in result.stderr, result
    with ProcessPoolExecutor(max_workers=1, mp_context=mp.get_context("spawn")) as pool:
        future = pool.submit(count_chunk, (8, 2))
        try:
            future.result(timeout=10)
        except ValueError as exc:
            assert str(exc) == "invalid interval", exc
        else:
            raise AssertionError("worker failure was lost")
    print("PASS: serial/process/one-worker agree; bad arguments rejected; worker error received")


if __name__ == "__main__":
    main()

执行：

python3 verify_pool.py

预期输出，退出码为 0：

PASS: serial/process/one-worker agree; bad arguments rejected; worker error received

测试给工作函数传入反向区间 (8, 2)，确认主进程通过 future.result() 收到 ValueError，而不是悄悄漏掉一个任务。这里的 timeout=10 只限制取结果的等待，不能终止已运行的任务；验证代码的子命令超时也不等于完整进程树管理。

6. 排错：按错误定位代码位置

现象	具体修复
`An attempt has been made to start a new process...`	把创建进程池的调用移入 `main()`，只在入口保护下调用；检查导入模块是否也在顶层启动任务。
`Can't pickle local object` 或 lambda 的 pickle 错误	把工作函数移到可导入模块顶层；不要使用嵌套函数或 lambda，传入整数、字符串或普通元组。
`Can't get attribute`、`<stdin>` 或 Notebook 中失败	保存为真实 `.py` 文件从终端运行；确认子进程能导入工作函数所在模块。
`BrokenProcessPool`	检查工作进程的真实回溯、内存限制或原生库崩溃；不要把丢失结果改成 0 继续成功。
更多进程反而更慢	先减少进程数、增加每块计算量，测完整命令的耗时；不要给每个整数单独提交一个任务。

官方 pickle 规则说明函数通过模块和名称序列化，而不是把函数代码随任务发送。本例不读取外部 pickle 文件；迁移真实任务时，也不要通过加载陌生 pickle 文件来“修复”传参。

如果任务主要是在调用外部命令，而不是做 Python 计算，可改用带日志和超时的 subprocess 调用。不要在工作函数里等待同一个进程池的新任务，也不要把数据库连接或打开的文件对象直接当参数传给它。

7. 总结

先保证子进程可以安全导入脚本，再考虑并行：顶层工作函数、受保护的入口和显式上下文分别解决不同问题。用同一输入比较串行与进程池结果，并真正取出 Future 的结果以暴露异常。只有输出正确且完整命令的实际耗时值得，才把进程池接入日常自动化。