Python中如何每30秒检测多进程是否存活？代码异常排查

你的代码核心问题出在**process.join()**这一行：join()方法会强制主进程阻塞，直到所有子进程完全运行结束才会继续执行后续代码。这就导致你的while循环只能在两个子进程都终止后才会启动，完全达不到“进程存活期间每30秒检查状态”的目的。

修改后的代码 #

import time
from multiprocessing import Process

def target1():
    # 模拟5-10分钟的业务逻辑
    time.sleep(300)  # 示例：运行5分钟

def target2():
    # 模拟5-10分钟的业务逻辑
    time.sleep(400)  # 示例：运行约6分40秒

if __name__ == "__main__":
    processes = []
    proc_1 = Process(target=target1)
    proc_1.start()
    processes.append(proc_1)

    proc_2 = Process(target=target2)
    proc_2.start()
    processes.append(proc_2)

    elapsed_time = 0
    process_timeout = 600  # 自定义超时时间，比如10分钟=600秒
    while elapsed_time < process_timeout:
        # 批量检查所有进程状态，扩展性更强
        is_any_alive = any(proc.is_alive() for proc in processes)
        if is_any_alive:
            print(f"进程仍在运行，已耗时{elapsed_time}秒，30秒后再次检查")
            time.sleep(30)
            elapsed_time += 30
        else:
            print("所有进程已正常结束")
            break
    else:
        # 循环未触发break，说明超时
        print(f"超时{process_timeout}秒，强制终止剩余存活进程")
        for proc in processes:
            if proc.is_alive():
                proc.terminate()
                proc.join()

关键调整说明 #

• 移除了阻塞主进程的join()循环，让检查逻辑能在子进程运行时同步执行
• 使用any()批量检查进程状态，后续新增进程也不用修改检查代码
• 增加了超时处理逻辑，可根据需求调整超时时间或移除该分支
• 补充了if __name__ == "__main__":包裹，适配Windows环境下的多进程运行要求

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