Python多进程Pool触发KeyboardInterrupt后获取未完成进程计算结果

问题核心原因 #

默认multiprocessing.Pool收到KeyboardInterrupt中断信号时，会直接发送终止信号杀死所有子进程，子进程没有机会执行收尾逻辑返回中间计算结果，因此仅在外层包裹try-except无法获取已生成的部分数据。

解决思路 #

使用跨进程共享的Event作为退出标记，子进程每轮迭代时检查标记状态，一旦标记被触发就立刻返回当前已计算的列表；主进程捕获中断信号后先触发退出标记，等待所有子进程返回结果后再清理进程池。

基础Demo修复版 #

可直接运行验证效果，触发Ctrl+C即可拿到子进程的中间结果：

from multiprocessing import Pool, Manager
from time import sleep

# 子进程全局变量，初始化时会注入共享退出标记
exit_event = None

def init_event(event):
    global exit_event
    exit_event = event

def fun(_):
    lst = []
    for i in range(200):
        # 每轮迭代先检查是否需要提前退出
        if exit_event.is_set():
            return lst
        lst.append(i)
        if not i % 10:
            sleep(0.1)
    return lst

if __name__ == '__main__':
    master = []
    processes = 2
    with Manager() as manager:
        # 创建跨进程共享的退出标记
        exit_event = manager.Event()
        # 初始化进程池时给所有子进程注入退出标记
        pool = Pool(
            processes=processes,
            initializer=init_event,
            initargs=(exit_event,)
        )
        try:
            for result in pool.imap_unordered(fun, range(processes)):
                master.append(result)
        except KeyboardInterrupt:
            print("\n收到中断信号，通知所有子进程返回中间结果")
            # 触发退出标记，通知所有子进程停止计算
            exit_event.set()
            # 关闭进程池入口，等待所有子进程执行完毕并返回结果
            pool.close()
            pool.join()
    # 输出所有收集到的结果（包含提前退出的子进程的中间结果）
    print(master)

适配你的实际业务代码 #

修改后的main.py #

from multiprocessing import Pool, Manager
from other import Other

class Something:
    def __init__(self, processes=2):
        self.processes = processes
        # 可自行根据业务调整参数
        self.episodes = 200
    
    def init_event(self, event):
        # 给子进程注入共享退出标记，同时传给Other类复用
        global exit_event
        exit_event = event
        Other.exit_event = event

    def spawner(self):
        master = []
        with Manager() as manager:
            exit_event = manager.Event()
            pool = Pool(
                processes=self.processes,
                initializer=self.init_event,
                initargs=(exit_event,)
            )
            try:
                for result in pool.imap_unordered(self.loop, range(self.processes)):
                    master.append(result)
                    # 此处写你原本的结果处理逻辑
            except KeyboardInterrupt:
                print("\n收到中断信号，收集子进程中间结果")
                exit_event.set()
                pool.close()
                pool.join()
        print("收集完成，最终结果：", master)
        return master

    def loop(self, _):
        # 此处保留你原本的setup逻辑
        setup_stuff = "你的业务配置"
        return Other(setup_stuff).start()

修改后的other.py #

class Other:
    # 退出标记会在子进程初始化时被赋值
    exit_event = None
    def __init__(self, setup_stuff):
        self.setup_stuff = setup_stuff
        # 可自行调整episodes参数来源
        self.episodes = 200

    def start(self):
        lst1, lst2, lst3 = [], [], []
        for _ in range(self.episodes):
            # 每轮迭代先检查是否需要提前退出
            if self.exit_event.is_set():
                return lst1, lst2, lst3
            # 此处保留你原本的计算逻辑，往三个列表中追加数据
            lst1.append(_)
            lst2.append(_ * 2)
            lst3.append(_ * 3)
        return lst1, lst2, lst3

注意事项 #

• 如果单次循环迭代耗时很长，可以在迭代内部多插入几个exit_event.is_set()检查点，提升中断响应速度
• 不要使用pool.terminate()方法，该方法会直接杀死子进程，不会等待子进程返回结果，本方案使用pool.close()+pool.join()等待子进程主动返回
• 该方案没有频繁的跨进程通信，仅在子进程终止时一次性返回结果，性能损耗极低

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