Python多进程：进程执行超时如何终止并让其余进程继续？

阿华AIGC实验室

2026-5-6

嘿，这个问题我之前刚踩过坑！官方文档确实有点太偏向理论，对初学者不太友好，咱们直接上解决方案，一步步来搞定它。

你的核心问题在于用了pool.map()——这个方法是阻塞式的，它会硬生生等所有子进程完成才会返回结果，哪怕某个进程像你示例里那样直接sys.exit()退出，或者卡住不动，父进程都会一直挂着等，完全没辙。要实现「单个进程超时就终止，其他进程继续跑」的需求，咱们得换个姿势用多进程。

方案一：用Pool的apply_async实现超时检查（简洁版）

如果只是需要标记超时/异常的任务，不需要强制杀掉进程，用apply_async就足够了。它是非阻塞的提交任务，咱们可以逐个给任务设置超时，单独处理异常情况：

import multiprocessing as mp
import sys
import time

def foo(x):
    if x == 3:
        sys.exit()  # 模拟子进程异常退出
    time.sleep(1)  # 模拟耗时计算
    return f"Result from task {x}"

if __name__ == '__main__':
    pool = mp.Pool(mp.cpu_count())
    task_list = []
    
    # 逐个提交任务，拿到每个任务的AsyncResult对象
    for num in [1, 2, 3]:
        task = pool.apply_async(foo, args=(num,))
        task_list.append((num, task))
    
    final_results = []
    for num, task in task_list:
        try:
            # 给每个任务设置2秒超时，超时就抛出TimeoutError
            result = task.get(timeout=2)
            final_results.append(result)
        except mp.TimeoutError:
            print(f"⚠️ Task {num} timed out, marking as failed")
            final_results.append(f"Task {num} failed (timeout)")
        except Exception as e:
            # 捕获子进程异常退出的情况
            print(f"❌ Task {num} hit an error: {str(e)}")
            final_results.append(f"Task {num} failed (error)")
    
    # 关闭池并等待所有剩余进程完成
    pool.close()
    pool.join()
    print("\nFinal Results:", final_results)

这个方案的特点：

代码简洁，不用手动管理进程，Pool会帮你复用进程资源
每个任务的超时检查是独立的，一个任务失败/超时不会影响其他任务
子进程异常退出时，Pool会自动重启新进程补位，不影响后续任务

方案二：手动管理进程实现强制终止（硬核版）

如果你必须强制杀掉超时的进程（比如某些任务卡住后会占着资源），那Pool就不太够用了——因为它封装了进程管理，没法直接操作单个进程。这时候咱们手动创建Process对象，用队列传递结果：

import multiprocessing as mp
import sys
import time

def foo(x, result_queue):
    try:
        if x == 3:
            sys.exit()  # 模拟异常退出
        time.sleep(1)
        # 把结果放进队列（进程间不能直接return值）
        result_queue.put((x, f"Success: Result from task {x}"))
    except Exception as e:
        result_queue.put((x, f"Error: {str(e)}"))

if __name__ == '__main__':
    processes = []
    result_queue = mp.Queue()
    timeout_seconds = 2
    
    # 创建并启动每个进程
    for num in [1, 2, 3]:
        p = mp.Process(target=foo, args=(num, result_queue))
        processes.append((num, p))
        p.start()
    
    # 逐个检查进程是否超时
    for num, p in processes:
        # 等待进程完成，最多等timeout_seconds秒
        p.join(timeout_seconds)
        if p.is_alive():
            print(f"🔪 Process {num} timed out, terminating it")
            p.terminate()  # 强制杀掉进程
            p.join()  # 确保进程彻底退出
            result_queue.put((num, "Failed: Task timed out"))
    
    # 从队列里收集所有结果
    final_results = []
    while not result_queue.empty():
        num, res = result_queue.get()
        final_results.append((num, res))
    
    # 按任务编号排序输出
    print("\nFinal Results:")
    for item in sorted(final_results):
        print(f"Task {item[0]}: {item[1]}")