Async executor执行后未终止进程：进程池复用与上限管控咨询

嘿，我来帮你理清楚这个问题！首先得明确：ProcessPoolExecutor的默认行为就是保留进程以实现复用，这其实是它的设计初衷——毕竟创建销毁进程的开销不小，复用能大幅提升性能。不过如果需要在合适的时机终止进程，我们也有明确的处理方式。

一、关于进程复用：你当前的代码已经在做了！ #

你初始化ProcessPoolExecutor(4)时，它会预先创建4个worker进程，这些进程会一直存活，等待处理后续通过run_in_executor提交的expensive任务。每次提交新任务，池子里的空闲进程会立刻接手，不会重新创建新进程，数量始终严格维持在你设定的4个上限内。这完全符合你“复用进程池+限制数量”的需求，不用做额外修改～

二、如何主动终止进程（比如应用 shutdown 时） #

如果你的问题是“应用退出后进程还残留”，那是因为ProcessPoolExecutor不会自动关闭，需要你主动调用shutdown()方法：

• shutdown(wait=True)（默认）：会等待所有已提交的任务执行完成，然后优雅终止所有worker进程
• shutdown(wait=False)：会立即终止所有进程，不管任务是否完成（仅推荐紧急场景使用）

给你的代码加个清理逻辑就能解决：

import asyncio
from concurrent.futures.process import ProcessPoolExecutor

# 补充你代码里的任务函数示例
def regular():
    return "Regular async task done"

def expensive():
    # 模拟CPU密集型计算
    import time
    time.sleep(2)
    return "CPU-heavy task done"

class App:
    def __init__(self):
        self.process_pool = ProcessPoolExecutor(4)
        self.loop = asyncio.get_event_loop()

    async def get_regular(self):
        return await regular()

    async def get_expensive(self):
        return await self.loop.run_in_executor(
            self.process_pool,
            expensive
        )
    
    def cleanup(self):
        # 优雅关闭进程池，等待所有任务完成后终止进程
        self.process_pool.shutdown(wait=True)
        print("Process pool closed, all workers terminated")

async def main():
    app = App()
    # 测试提交多个任务，验证进程复用
    task1 = asyncio.create_task(app.get_expensive())
    task2 = asyncio.create_task(app.get_expensive())
    results = await asyncio.gather(task1, task2)
    print(results)
    
    # 应用结束前清理资源
    app.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

几个关键注意点 #

• 长期运行的异步应用：保持进程池存活复用是最优解，不用频繁关闭重启，避免进程创建的额外开销
• 短期任务或脚本：一定要在应用退出前调用shutdown()，否则worker进程会成为“僵尸进程”残留
• 不要在每次任务提交后创建新的ProcessPoolExecutor，这会完全失去进程复用的意义，性能反而更差

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