如何维护进程池？进程崩溃时自动补充直至任务队列清空

完全可行，核心思路是主动监控进程状态，一旦进程退出（无论正常完成还是异常崩溃），就根据队列是否还有任务决定是否重启新进程，直到任务队列清空且所有进程都正常退出。

实现步骤与代码示例 #

1. 任务处理函数 #

定义任务处理函数run_alg，从队列中获取任务并处理，遇到异常时直接退出进程，让主进程检测到并重启：

import multiprocessing
import time

def run_alg(queue):
    while True:
        try:
            # 带超时获取任务，避免进程一直阻塞在空队列上
            task = queue.get(block=True, timeout=1)
            print(f"进程 {multiprocessing.current_process().pid} 处理任务: {task}")
            
            # 模拟任务处理，加入崩溃测试逻辑
            if task % 5 == 0:
                raise RuntimeError("模拟进程崩溃")
            time.sleep(0.5)
            
        except multiprocessing.queues.Empty:
            # 超时后二次检查队列是否真的为空，为空则退出进程
            if queue.empty():
                break
            continue
        except Exception as e:
            print(f"进程 {multiprocessing.current_process().pid} 崩溃，错误: {str(e)}")
            break  # 崩溃后直接退出进程

2. 主进程的进程监控与重启逻辑 #

主进程负责初始化进程池、填充任务队列，并持续监控进程状态，自动重启退出的进程：

def main():
    target_process_num = 3  # 需要维持的进程数量
    queue = multiprocessing.Queue()

    # 填充任务队列（示例：20个任务）
    for task_id in range(20):
        queue.put(task_id)

    # 初始化初始进程池
    processes = []
    for _ in range(target_process_num):
        p = multiprocessing.Process(target=run_alg, args=(queue,))
        p.start()
        processes.append(p)
        print(f"启动初始进程 {p.pid}")

    # 持续监控进程状态，直到任务全部完成
    while True:
        alive_processes = []
        for p in processes:
            if p.is_alive():
                alive_processes.append(p)
            else:
                # 进程已退出，检查队列是否还有任务
                if not queue.empty():
                    # 重启新进程替代已退出的进程
                    new_p = multiprocessing.Process(target=run_alg, args=(queue,))
                    new_p.start()
                    alive_processes.append(new_p)
                    print(f"进程 {p.pid} 退出，重启新进程 {new_p.pid}")
                # 队列已空则不再重启

        # 更新存活进程列表
        processes = alive_processes

        # 结束条件：队列空且无存活进程
        if not processes and queue.empty():
            break

        # 降低检查频率，减少资源消耗
        time.sleep(0.5)

    print("所有任务处理完成")

if __name__ == "__main__":
    main()

关键说明 #

• 进程状态监控：通过is_alive()方法实时检查进程存活状态，退出的进程会被自动替换（仅当队列还有任务时）。
• 队列空判断：任务处理函数中通过超时+二次检查队列状态，避免进程在任务未完成时提前退出。
• 崩溃容错：任务处理函数捕获异常后直接退出，主进程会立即重启新进程，确保始终维持指定数量的工作进程处理任务。

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