subprocess 如何在 timeout 时优雅杀死子进程树

12次阅读

subprocess.run() 的 timeout 参数只终止主进程,不清理子进程树;应使用 start_new_session=True(linux/macos)或 CREATE_NEW_PROCESS_GROUP(windows)配合 killpg/CTRL_BREAK_EVENT,或用 psutil 显式遍历终止整个进程树。

subprocess 如何在 timeout 时优雅杀死子进程树

subprocess.run() 的 timeout 参数只杀主进程,不清理子进程树

pythonsubprocess.run()subprocess.Popen.wait(timeout=...) 在超时时只会向直接子进程发送信号(如 SIGTERM),但不会递归终止其派生的整个进程树。这意味着 shell 启动的命令链(如 bash -c "sleep 10 & sleep 20")或通过 nohup/& 后台启动的子进程大概率残留。

用 start_new_session=True 配合 os.killpg() 杀整个进程组

关键在于让子进程及其后代运行在独立会话(session)中,这样就能用进程组 ID(PGID)一次性终结整棵树。核心是:start_new_session=True(Linux/macOS)或 creationflags=subprocess.CREATE_NEW_PROCESS_GROUPwindows)。

  • Linux/macOS 下:调用 os.setsid() 创建新会话,主进程成为会话首进程,其 PGID = PID;后续 fork 出的子进程默认加入该 PGID
  • 启动时必须加 start_new_session=True,否则 os.getpgid(proc.pid) 可能报错或返回父进程组 ID
  • 超时后用 os.killpg(os.getpgid(proc.pid), signal.SIGTERM) 终止整个组,再 wait() 收尸
import subprocess, os, signal, time 

proc = subprocess.Popen( ["bash", "-c", "sleep 5; echo 'done'"], start_newsession=True, # 必须! stdout=subprocess.pipE, stderr=subprocess.STDOUT ) try: stdout, = proc.communicate(timeout=2) except subprocess.TimeoutExpired: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) proc.wait() # 等待进程组彻底退出

Windows 上要用 CREATE_NEW_PROCESS_GROUP + CTRL_BREAK_EVENT

Windows 没有 POSIX 进程组概念,os.killpg 不可用。必须用 subprocess.CREATE_NEW_PROCESS_GROUP 创建独立进程组,再用 os.kill() 发送 signal.CTRL_BREAK_EVENT(不能用 CTRL_C_EVENT,它可能被目标进程忽略)。

  • CTRL_BREAK_EVENT 能传递给整个控制台进程组,比 TerminateProcess 更温和
  • 必须确保子进程是控制台应用(非 GUI),否则信号无效
  • 调用前需先 proc.terminate() 或直接 os.kill(proc.pid, signal.CTRL_BREAK_EVENT),之后仍要 proc.wait()

更健壮的做法:用 psutil 回收所有后代进程

如果无法控制启动参数(比如不能加 start_new_session),或者跨平台兼容性要求高,推荐用 psutil 手动遍历并终止子树。它不依赖会话机制,而是靠 proc.children(recursive=True) 获取全部后代。

  • 安装:pip install psutil
  • 注意:Windows 上需管理员权限才能终止某些系统相关子进程
  • 务必按逆序终止(先叶子后父进程),避免子进程被 init 进程领养而逃逸
  • 示例中 proc.children(recursive=True) 返回的是实时快照,若进程瞬间启停,可能漏掉极短命子进程
import subprocess, psutil, time 

proc = subprocess.Popen(["bash", "-c", "sleep 3 &"]) try: proc.communicate(timeout=1) except subprocess.TimeoutExpired: parent = psutil.Process(proc.pid) children = parent.children(recursive=True) for child in reversed(children): # 先杀孙子,再杀儿子 try: child.terminate() except (psutil.NoSuchProcess, psutil.accessDenied): pass try: parent.terminate() parent.wait(timeout=3) except (psutil.NoSuchProcess, psutil.TimeoutExpired): pass

实际项目里,start_new_session=True 是最轻量且可靠的选择,但前提是能掌控子进程启动方式;一旦涉及遗留脚本、第三方二进制或容器化环境,psutil 的显式树遍历反而更可控——毕竟进程关系不是靠约定,而是靠实时探测。

text=ZqhQzanResources