下面是运行任意命令返回其标准输出数据的Python代码，或在非零退出码上引发异常:
proc = subprocess.Popen(
cmd,
stderr=subprocess.STDOUT, # Merge stdout and stderr
stdout=subprocess.PIPE,
shell=True)
communication用于等待进程退出:
stdoutdata, stderrdata = proc.communicate()
子进程模块不支持超时——杀死运行超过X秒的进程的能力——因此，通信可能需要很长时间才能运行。
在Windows和Linux上运行的Python程序中实现超时的最简单方法是什么?
当前回答
有时需要处理(ffmpeg)而不使用communication()，在这种情况下需要异步超时，这是使用ttldict实现的一种实用方法
PIP安装ttldict
from ttldict import TTLOrderedDict
sp_timeout = TTLOrderedDict(default_ttl=10)
def kill_on_timeout(done, proc):
while True:
now = time.time()
if sp_timeout.get('exp_time') == None:
proc.kill()
break
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, stderr=subprocess.STDOUT)
sp_timeout['exp_time'] = time.time()
done = Event()
watcher = Thread(target=kill_on_timeout, args=(done, process))
watcher.daemon = True
watcher.start()
done.set()
for line in process.stdout:
.......
其他回答
这是我的解决方案，我使用线程和事件:
import subprocess
from threading import Thread, Event
def kill_on_timeout(done, timeout, proc):
if not done.wait(timeout):
proc.kill()
def exec_command(command, timeout):
done = Event()
proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
watcher = Thread(target=kill_on_timeout, args=(done, timeout, proc))
watcher.daemon = True
watcher.start()
data, stderr = proc.communicate()
done.set()
return data, stderr, proc.returncode
在行动:
In [2]: exec_command(['sleep', '10'], 5)
Out[2]: ('', '', -9)
In [3]: exec_command(['sleep', '10'], 11)
Out[3]: ('', '', 0)
预先设置Linux命令超时并不是一个糟糕的解决方法，它对我来说是有效的。
cmd = "timeout 20 "+ cmd
subprocess.Popen(cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(output, err) = p.communicate()
在Python 3.7.8中超时测试后捕获的输出示例:
try:
return subprocess.run(command, shell=True, capture_output=True, timeout=20, cwd=cwd, universal_newlines=True)
except subprocess.TimeoutExpired as e:
print(e.output.decode(encoding="utf-8", errors="ignore"))
assert False;
异常子进程。TimeoutExpired有输出和其他成员:
cmd -用于生成子进程的命令。 timeout -超时时间，单位为秒。 output -子进程的输出，如果它是由run()或 check_output()。否则,没有。 stdout -输出的别名，用于与stderr对称。 stderr -子进程的stderr输出，如果它被捕获 run()。否则,没有。
更多信息:https://docs.python.org/3/library/subprocess.html#subprocess.TimeoutExpired
我使用的解决方案是在shell命令前面加上timelimit。如果命令花费的时间太长，timelimit将停止它，Popen将有一个由timelimit设置的返回码。如果它是> 128，它意味着时间限制杀死进程。
参见python subprocess with timeout and large output (>64K)
This solution kills the process tree in case of shell=True, passes parameters to the process (or not), has a timeout and gets the stdout, stderr and process output of the call back (it uses psutil for the kill_proc_tree). This was based on several solutions posted in SO including jcollado's. Posting in response to comments by Anson and jradice in jcollado's answer. Tested in Windows Srvr 2012 and Ubuntu 14.04. Please note that for Ubuntu you need to change the parent.children(...) call to parent.get_children(...).
def kill_proc_tree(pid, including_parent=True):
parent = psutil.Process(pid)
children = parent.children(recursive=True)
for child in children:
child.kill()
psutil.wait_procs(children, timeout=5)
if including_parent:
parent.kill()
parent.wait(5)
def run_with_timeout(cmd, current_dir, cmd_parms, timeout):
def target():
process = subprocess.Popen(cmd, cwd=current_dir, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
# wait for the process to terminate
if (cmd_parms == ""):
out, err = process.communicate()
else:
out, err = process.communicate(cmd_parms)
errcode = process.returncode
thread = Thread(target=target)
thread.start()
thread.join(timeout)
if thread.is_alive():
me = os.getpid()
kill_proc_tree(me, including_parent=False)
thread.join()
