下面是运行任意命令返回其标准输出数据的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程序中实现超时的最简单方法是什么?
当前回答
另一种选择是写入临时文件以防止标准输出阻塞，而不需要使用communication()轮询。在其他答案没有的地方，这个方法对我有用;比如在窗户上。
outFile = tempfile.SpooledTemporaryFile()
errFile = tempfile.SpooledTemporaryFile()
proc = subprocess.Popen(args, stderr=errFile, stdout=outFile, universal_newlines=False)
wait_remaining_sec = timeout
while proc.poll() is None and wait_remaining_sec > 0:
time.sleep(1)
wait_remaining_sec -= 1
if wait_remaining_sec <= 0:
killProc(proc.pid)
raise ProcessIncompleteError(proc, timeout)
# read temp streams from start
outFile.seek(0);
errFile.seek(0);
out = outFile.read()
err = errFile.read()
outFile.close()
errFile.close()
其他回答
这是我的解决方案，我使用线程和事件:
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)
在Python 3.3+中:
from subprocess import STDOUT, check_output
output = check_output(cmd, stderr=STDOUT, timeout=seconds)
输出是一个字节字符串，包含命令合并的标准输出、标准输出数据。
与proc. communication()方法不同，check_output在问题文本中指定的非零退出状态上引发CalledProcessError。
我删除了shell=True，因为它经常被不必要地使用。如果cmd确实需要它，您总是可以将它添加回来。如果你添加shell=True，也就是说，如果子进程产生了自己的后代;check_output()的返回时间可能比超时时间晚得多，请参见子进程超时失败。
超时特性在Python 2中可用。X通过3.2+ subprocess模块的subprocess32后端端口实现。
您可以使用select来完成此操作
import subprocess
from datetime import datetime
from select import select
def call_with_timeout(cmd, timeout):
started = datetime.now()
sp = subprocess.Popen(cmd, stdout=subprocess.PIPE)
while True:
p = select([sp.stdout], [], [], timeout)
if p[0]:
p[0][0].read()
ret = sp.poll()
if ret is not None:
return ret
if (datetime.now()-started).total_seconds() > timeout:
sp.kill()
return None
另一种选择是写入临时文件以防止标准输出阻塞，而不需要使用communication()轮询。在其他答案没有的地方，这个方法对我有用;比如在窗户上。
outFile = tempfile.SpooledTemporaryFile()
errFile = tempfile.SpooledTemporaryFile()
proc = subprocess.Popen(args, stderr=errFile, stdout=outFile, universal_newlines=False)
wait_remaining_sec = timeout
while proc.poll() is None and wait_remaining_sec > 0:
time.sleep(1)
wait_remaining_sec -= 1
if wait_remaining_sec <= 0:
killProc(proc.pid)
raise ProcessIncompleteError(proc, timeout)
# read temp streams from start
outFile.seek(0);
errFile.seek(0);
out = outFile.read()
err = errFile.read()
outFile.close()
errFile.close()
Jcollado的答案可以使用线程来简化。定时器类:
import shlex
from subprocess import Popen, PIPE
from threading import Timer
def run(cmd, timeout_sec):
proc = Popen(shlex.split(cmd), stdout=PIPE, stderr=PIPE)
timer = Timer(timeout_sec, proc.kill)
try:
timer.start()
stdout, stderr = proc.communicate()
finally:
timer.cancel()
# Examples: both take 1 second
run("sleep 1", 5) # process ends normally at 1 second
run("sleep 5", 1) # timeout happens at 1 second
