Python multiprocessing.Pool 停机后进程仍不退：close、terminate、join 顺序排查

批处理服务收到停止信号后，主进程日志已经输出「开始退出」，但CPU占用和子进程还一直在占用资源，最常见的原因就是 multiprocessing.Pool 的收尾顺序写反了。close()、terminate()、join() 不是三个同义的“关闭方法”，它们对应的是三种完全不同的生命周期动作。

先复现“主流程结束，worker还在跑”的问题

下面的例子模拟图片缩略图批量处理场景。每个worker进程负责处理一个文件，主进程提交完所有任务就进入退出流程。如果只写 join()，程序会在部分场景下直接报错，或是永远等不到合理的收尾节点，因为Pool本身还没被告知要不要继续接收新任务。

from multiprocessing import Pool
import time

def make_thumbnail(path: str) -> str:
    time.sleep(0.2)
    return f"done:{path}"

pool = Pool(processes=2)
results = [pool.apply_async(make_thumbnail, (f"img-{i}.jpg",)) for i in range(4)]

# 错误示例：只等待，不声明 Pool 的下一状态
pool.join()

这段代码的核心问题不是“进程池运行慢”，而是进程池的生命周期没有完整闭合。主进程必须先明确自己的意图：是等已提交的任务全部做完，还是立刻放弃所有没跑完的任务。确定好选择之后，join() 才有明确的等待对象，不会出问题。

close、terminate、join 分别负责什么

可以把Pool理解成一个自带任务输入口和一批worker进程的批处理器。三个方法的职责划分得很清楚：

所以正常走完所有任务的流程是 close() -> join()，要强制放弃所有任务的流程是 terminate() -> join()。如果把 join() 放在最前面，相当于还没决定进程池是要收完任务再停还是立刻停，排查的时候很容易陷入找不到原因的无限等待。

正常批处理应该先把所有结果消费完

如果每个任务的执行结果都不能丢，可以直接用 map() 或者拿到所有异步任务的 AsyncResult 之后逐个读取返回值。读取结果的时候，worker内部抛出的异常会在主进程里重新抛出来，不能只检查进程池是不是退出了就完事。

from multiprocessing import Pool

def build_report(day: str) -> str:
    if day == "bad-input":
        raise ValueError("invalid report date")
    return f"report:{day}"

pool = Pool(processes=2)
try:
    jobs = [pool.apply_async(build_report, (day,)) for day in ["2026-07-25", "bad-input"]]
    pool.close()
    for job in jobs:
        print(job.get(timeout=5))
except Exception:
    pool.terminate()
    raise
else:
    pool.join()

这里有两个很关键的检查点。第一，close() 要放在所有任务都提交完成之后，避免后续代码误加新任务进去。第二，要通过 get() 读取每个worker的执行结果，才能把 ValueError 这类业务错误传回主进程；只看到worker进程的数量下降，根本不能证明整批任务处理成功了。

异常和超时路径必须主动切断未完成任务

批处理最麻烦的情况就是某一个worker卡在外部I/O操作，其他任务都跑完了，但主进程还在无限等待。这时候不要靠无限拉长 join() 的等待时长来解决问题，应该给单个任务结果、整个停机流程都设置明确的最大等待上限。

import logging
from multiprocessing import Pool

logger = logging.getLogger("thumbnail-batch")

def run_batch(paths: list[str]) -> list[str]:
    pool = Pool(processes=2)
    jobs = []
    try:
        jobs = [pool.apply_async(make_thumbnail, (path,)) for path in paths]
        pool.close()
        values = [job.get(timeout=10) for job in jobs]
        pool.join()
        return values
    except Exception:
        logger.exception("batch failed; terminating workers")
        pool.terminate()
        pool.join()
        raise

调用完 terminate() 之后还是要补调用 join()。前者只是发出了停止子进程的动作，后者才会确认子进程真的完全退出、系统资源被回收。少了第二步的话，就算日志里已经打印了“终止进程池”的提示，操作系统的进程列表里还是可能短时间残留这些worker进程。

就算用with Pool也不等于业务逻辑绝对安全

with Pool(...) as pool 这种上下文管理器写法可以帮你自动执行收尾操作，但它不会替你做业务决策要不要放弃未完成的任务。退出上下文的时候，Pool会走默认的终止式清理逻辑；如果这批处理的结果必须全部落库、必须生成完所有目标文件，最好显式把所有结果都消费完、确认业务逻辑成功之后，再离开上下文管理器的作用域。

from multiprocessing import Pool

def export_all(paths: list[str]) -> list[str]:
    with Pool(processes=2) as pool:
        jobs = [pool.apply_async(make_thumbnail, (path,)) for path in paths]
        values = [job.get(timeout=10) for job in jobs]
        return values

files = export_all(["a.jpg", "b.jpg"])
print(files)

这个写法很适合短任务、已经明确结果读取逻辑的场景。如果进程收到外部的停机信号，你需要提前定义服务侧的处理策略：允许跑完当前批次就走 close() 和 join() 的流程，不允许就先记录未完成的任务清单再终止进程池。

上线前用四个场景核对收尾逻辑

1. 全部任务成功：确认所有 get() 都能正常返回，走完 close() 之后 join() 能在预留的停机时间预算内完成。
2. 单个任务报错：确认主进程能拿到worker抛出的原始异常，其他未完成的任务会按预设的策略执行完或者直接终止。
3. 单个任务超时：确认日志里记录了任务的输入参数、超时秒数和终止原因，而不是只写一句“任务卡住”的模糊提示。
4. 收到停机信号：确认不会再接收提交新的任务，所有worker进程最终数量归零，临时文件和中间运行状态都可以回溯追踪。

如果你的批处理完全不能接受进程终止带来的数据丢失，那就需要在任务启动前先存好可重试的输入清单，下一次服务启动的时候直接从清单里恢复执行，不要把Pool本身当成持久化队列来用。

常见问题

close() 会马上杀掉worker进程吗？

不会。它只做禁止新任务进入的动作，已经提交的任务还是会继续执行完；要等所有worker退出，需要再调用 join()。

terminate() 之后还要调用join()吗？

要。terminate() 只是发起停止动作，join() 负责等待进程退出并确认系统资源完全回收，两个动作不能混为一谈。

job.get(timeout=10) 超时之后还能继续用这个Pool吗？

可以，但你要先判断这个卡住的任务能不能安全继续运行。对无法确定状态的外部I/O场景，更稳妥的做法是先记录任务输入，直接终止当前Pool，之后用幂等的方式重试这个任务。

为什么worker里抛了错，主进程却没有退出？

异步结果对象会把worker里的异常存下来，只有主进程调用 get() 等方法主动读取结果的时候，异常才会在主进程里重新抛出。要把异常处理纳入控制流，不能只等进程池自动结束。

把Pool的两条退出路径写清楚

这类故障的判断逻辑很简单：正常跑完所有任务就用 close() -> join()，遇到异常、超时或是明确要放弃任务的场景就用 terminate() -> join()。再把 get() 加到结果核对的环节里，主进程日志、worker运行状态和业务处理结果才能完全对应上。