下面是运行任意命令返回其标准输出数据的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程序中实现超时的最简单方法是什么?
当前回答
我已经在Windows, Linux和Mac上成功地使用了killableprocess。如果你使用Cygwin Python，你将需要OSAF的killableprocess版本，否则本机Windows进程将不会被杀死。
其他回答
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()
如果你用的是Unix，
import signal
...
class Alarm(Exception):
pass
def alarm_handler(signum, frame):
raise Alarm
signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm(5*60) # 5 minutes
try:
stdoutdata, stderrdata = proc.communicate()
signal.alarm(0) # reset the alarm
except Alarm:
print "Oops, taking too long!"
# whatever else
有时需要处理(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:
.......
虽然我还没有广泛地研究它，但我在ActiveState中发现的这个装饰器似乎对这类事情非常有用。伴随着subprocess.Popen(…， close_fds=True)，至少我已经准备好在Python中编写shell脚本。
从Python 3.5开始，有了一个新的subprocess.run通用命令(意味着取代check_call, check_output…)，它也有timeout=参数。
subprocess.run(args， *， stdin=None, input=None, stdout=None, stderr=None, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None) 执行args描述的命令。等待命令完成，然后返回一个CompletedProcess实例。
它引发一个子流程。超时超时时出现TimeoutExpired异常。
