如何让Python子进程处理Ctrl-C中断而非终止父进程?
我有一个作为包装器的Python文件parent.py,用于启动另一个Python程序child.py。parent.py的核心代码如下:
p = subprocess.run(f"python -u {cmd} 2>&1 | tee -a {file_path}", shell=True)
我希望使用Ctrl-C控制子进程,因此在child.py中编写了try-except逻辑来捕获KeyboardInterrupt并进行处理:
try: while True: pass except KeyboardInterrupt: print("lalala~~~~~~~~~~~~~~~~~~~~~")
但运行parent.py并按下Ctrl-C后,父进程被终止,子进程无法处理该中断(无"lalala"输出),报错回溯如下:
^CTraceback (most recent call last): File
"/home/zxl/rcd_python1.py", line 27, in
p = subprocess.run(f"python -u {cmd} 2>&1 | tee -a {file_path}", shell=True) File
"/home/zxl/programs/anaconda3/envs/pyg2/lib/python3.8/subprocess.py",
line 495, in run
stdout, stderr = process.communicate(input, timeout=timeout) File
"/home/zxl/programs/anaconda3/envs/pyg2/lib/python3.8/subprocess.py",
line 1020, in communicate
self.wait() File "/home/zxl/programs/anaconda3/envs/pyg2/lib/python3.8/subprocess.py",
line 1083, in wait
return self._wait(timeout=timeout) File "/home/zxl/programs/anaconda3/envs/pyg2/lib/python3.8/subprocess.py",
line 1808, in _wait
(pid, sts) = self._try_wait(0) File "/home/zxl/programs/anaconda3/envs/pyg2/lib/python3.8/subprocess.py",
line 1766, in _try_wait
(pid, sts) = os.waitpid(self.pid, wait_flags) KeyboardInterrupt
请问如何将Ctrl-C发送给子进程并让其处理,而非终止所有进程?
问题核心原因:使用shell=True时,subprocess会启动一个shell进程作为父进程的直接子进程,child.py是该shell的子进程。按下Ctrl-C时,操作系统会给整个前台进程组发送SIGINT信号,父Python进程、shell进程、子Python进程都会收到。父进程默认抛出KeyboardInterrupt终止,shell进程收到信号后也会终止,导致child.py来不及处理信号就被终止。
以下是几种可行的解决方法:
方法1:父进程捕获信号并转发给子进程
修改parent.py,捕获SIGINT后手动将信号发送给子进程,避免父进程直接终止:
import subprocess import signal import os def handle_sigint(signum, frame): if p.poll() is None: # 子进程仍在运行 os.kill(p.pid, signal.SIGINT) cmd = "child.py" file_path = "output.log" # 注册信号处理函数 signal.signal(signal.SIGINT, handle_sigint) p = subprocess.run(f"python -u {cmd} 2>&1 | tee -a {file_path}", shell=True)
注意:这种方式下shell进程收到SIGINT后可能仍会终止,child.py的输出可能无法被tee完整捕获。若需保留tee功能,可选用方法2。
方法2:不使用shell=True,手动实现tee功能
直接启动child.py,在父进程中处理输出同时写入文件,完全控制信号传递:
import subprocess import signal import sys def handle_sigint(signum, frame): if p.poll() is None: p.send_signal(signal.SIGINT) cmd = ["python", "-u", "child.py"] file_path = "output.log" signal.signal(signal.SIGINT, handle_sigint) # 启动子进程,捕获输出并同时写入控制台和文件 with open(file_path, "a") as f: p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) for line in p.stdout: sys.stdout.write(line) f.write(line) sys.stdout.flush() f.flush() p.wait()
这种方式下,子进程是父进程的直接子进程,父进程捕获SIGINT后转发给子进程,子进程可正常处理并输出"lalala",父进程会等待子进程处理完成后再退出。
方法3:子进程切换进程组(适用于必须用shell=True的场景)
让child.py启动后切换到新进程组,避免收到默认的SIGINT,再由父进程手动发送信号给子进程:
修改child.py:
import os import signal # 切换到新进程组,脱离默认信号传递范围 os.setpgid(0, 0) try: while True: pass except KeyboardInterrupt: print("lalala~~~~~~~~~~~~~~~~~~~~~")
修改parent.py(需安装psutil库:pip install psutil):
import subprocess import signal import os import psutil def get_child_pids(parent_pid): # 获取指定父进程下的所有子进程PID child_pids = [] try: parent = psutil.Process(parent_pid) child_pids = [child.pid for child in parent.children(recursive=True)] except psutil.NoSuchProcess: pass return child_pids def handle_sigint(signum, frame): if p.poll() is None: # 获取shell进程的子进程(即child.py的PID) child_pids = get_child_pids(p.pid) for pid in child_pids: os.kill(pid, signal.SIGINT) cmd = "child.py" file_path = "output.log" signal.signal(signal.SIGINT, handle_sigint) p = subprocess.run(f"python -u {cmd} 2>&1 | tee -a {file_path}", shell=True)
内容的提问来源于stack exchange,提问作者Jojo Zhai

