如何让multiprocessing Pool不启动新任务且不终止正在运行的进程？

原生multiprocessing.Pool提供的close()和terminate()确实无法满足你描述的「运行中任务继续完成、未启动任务不再执行」的需求，核心原因是close()仅会阻止后续新的任务提交到池，已经写入池内部待执行队列的任务依然会被worker拉取执行。可以通过以下两种方案实现预期效果：

方案1：逐批提交任务（推荐，无私有属性依赖） #

该方案不依赖Pool的内部实现，逻辑通用兼容性好，适合新开发的业务场景：

• 不要一开始就用map/apply_async把所有任务全部提交到Pool
• 维护全局待执行任务列表，提交任务前先校验是否在运行窗口内，窗口到期后立刻停止提交
• 控制已提交未完成的任务数量不要超过进程池大小，避免内部队列堆积多余任务
• 停止提交后调用Pool.close() + Pool.join()，此时正在运行的任务会正常跑完，不会有新任务启动

Python 2.7示例代码：

import multiprocessing
import time
from datetime import datetime

# 自定义单任务执行逻辑
def run_task(task_param):
    print("处理任务: %s" % task_param)
    time.sleep(10) # 模拟长耗时任务
    return task_param

# 自定义运行窗口判断逻辑，示例为每日1点到5点允许运行
def is_in_run_window():
    now_hour = datetime.now().hour
    return 1 <= now_hour < 5

if __name__ == "__main__":
    max_workers = 4 # 最多同时并行4个任务
    pool = multiprocessing.Pool(processes=max_workers)
    all_tasks = range(1000) # 所有待执行任务的参数列表
    uncompleted_handlers = []

    for task_param in all_tasks:
        # 提交前校验运行窗口，到期直接终止提交循环
        if not is_in_run_window():
            print("运行窗口到期，停止提交新任务")
            break
        # 提交单个任务
        res = pool.apply_async(run_task, args=(task_param,))
        uncompleted_handlers.append(res)
        # 控制队列不堆积，已提交未完成的任务不超过进程数+1，避免空等
        while len(uncompleted_handlers) >= max_workers + 1:
            # 清理已完成的任务句柄
            finished = [r for r in uncompleted_handlers if r.ready()]
            for r in finished:
                uncompleted_handlers.remove(r)
            time.sleep(1)

    # 等待所有已提交的任务执行完成
    pool.close()
    pool.join()
    print("所有已启动任务处理完成")

方案2：清空Pool内部待执行队列（适配已全量提交任务的老场景） #

如果你的代码已经实现了全量提交所有任务到Pool的逻辑，可以直接操作Python 2.7 Pool的内部私有属性_taskqueue清空未执行的任务，再调用close()即可实现预期效果。

注意：该方案依赖Pool的内部实现，仅适用于Python 2.7固定版本，跨版本可能存在兼容性问题

Python 2.7示例代码：

import multiprocessing
import time
from datetime import datetime

def run_task(task_param):
    print("处理任务: %s" % task_param)
    time.sleep(10)
    return task_param

def is_in_run_window():
    now_hour = datetime.now().hour
    return 1 <= now_hour < 5

if __name__ == "__main__":
    max_workers = 4
    pool = multiprocessing.Pool(processes=max_workers)
    all_tasks = range(1000)

    # 原有逻辑：全量提交所有任务到Pool
    for param in all_tasks:
        pool.apply_async(run_task, args=(param,))

    # 轮询检查运行窗口
    while is_in_run_window():
        # 所有任务执行完直接退出
        if pool._taskqueue.empty():
            break
        time.sleep(5)

    # 窗口到期，清空所有未被worker拉取的待执行任务
    while not pool._taskqueue.empty():
        try:
            pool._taskqueue.get()
        except:
            break

    # 关闭池，等待正在运行的任务全部完成
    pool.close()
    pool.join()
    print("处理完成")

内容的提问来源于stack exchange，提问作者WarSame