如何优雅地通知Python的multiprocessing.Pool中的进程退出？

import multiprocessing
import time
import signal
import sys

class GracefulExitPool:
    def __init__(self, processes=None):
        self.pool = multiprocessing.Pool(processes=processes)
        self.shutdown_event = multiprocessing.Event()
        
    def worker(self, task):
        """工作函数，定期检查退出信号"""
        if self.shutdown_event.is_set():
            print(f"进程 {multiprocessing.current_process().name} 收到退出信号")
            return None
            
        # 模拟工作
        time.sleep(0.5)
        return f"处理任务: {task}"
    
    def apply_async_with_check(self, task):
        """包装apply_async，添加退出检查"""
        return self.pool.apply_async(self._worker_wrapper, (task,))
    
    def _worker_wrapper(self, task):
        """包装器函数，用于进程内部"""
        if self.shutdown_event.is_set():
            return None
        return self.worker(task)
    
    def shutdown(self):
        """优雅关闭池"""
        print("发送关闭信号...")
        self.shutdown_event.set()
        
        # 等待当前任务完成
        self.pool.close()
        self.pool.join()
        print("所有进程已优雅退出")

# 使用示例
if __name__ == "__main__":
    pool = GracefulExitPool(processes=4)
    
    try:
        # 提交任务
        results = []
        for i in range(10):
            result = pool.apply_async_with_check(i)
            results.append(result)
        
        # 模拟运行一段时间后需要退出
        time.sleep(2)
        
        # 获取部分结果
        for r in results[:5]:
            print(r.get(timeout=1))
        
    except KeyboardInterrupt:
        print("\n收到中断信号")
    finally:
        # 优雅关闭
        pool.shutdown()

核心方法：

1. 使用Event信号：multiprocessing.Event()作为全局退出标志，所有工作进程定期检查这个标志
2. 包装工作函数：在真正的工作逻辑外包一层检查退出的包装器
3. close() + join()组合：先阻止新任务提交，再等待现有任务完成

关键点：

• 不要用terminate()，它会强制杀死进程，可能导致资源泄漏
• 工作函数中需要定期检查退出标志（比如在循环中或任务间隙）
• 配合try-finally确保无论是否异常都能执行清理

一句话总结：用Event信号配合close/join实现优雅退出。