下面是运行任意命令返回其标准输出数据的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程序中实现超时的最简单方法是什么?
当前回答
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
其他回答
如果你用的是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
如果你正在使用python2，请尝试一下
import subprocess32
try:
output = subprocess32.check_output(command, shell=True, timeout=3)
except subprocess32.TimeoutExpired as e:
print e
https://pypi.python.org/pypi/python-subprocess2提供了子流程模块的扩展，允许您等待一段时间，否则将终止。
因此，等待最多10秒的进程终止，否则kill:
pipe = subprocess.Popen('...')
timeout = 10
results = pipe.waitOrTerminate(timeout)
这与windows和unix兼容。"results"是一个字典，它包含"returnCode"，这是应用程序的返回(或None，如果它必须被杀死)，以及"actionTaken"。如果流程正常完成，则为“SUBPROCESS2_PROCESS_COMPLETED”，或者根据所采取的操作，为“SUBPROCESS2_PROCESS_TERMINATED”和SUBPROCESS2_PROCESS_KILLED(详细信息请参阅文档)
我已经实现了我能从其中一些收集到的东西。这适用于Windows，因为这是一个社区维基，我想我也会分享我的代码:
class Command(threading.Thread):
def __init__(self, cmd, outFile, errFile, timeout):
threading.Thread.__init__(self)
self.cmd = cmd
self.process = None
self.outFile = outFile
self.errFile = errFile
self.timed_out = False
self.timeout = timeout
def run(self):
self.process = subprocess.Popen(self.cmd, stdout = self.outFile, \
stderr = self.errFile)
while (self.process.poll() is None and self.timeout > 0):
time.sleep(1)
self.timeout -= 1
if not self.timeout > 0:
self.process.terminate()
self.timed_out = True
else:
self.timed_out = False
然后从另一个类或文件:
outFile = tempfile.SpooledTemporaryFile()
errFile = tempfile.SpooledTemporaryFile()
executor = command.Command(c, outFile, errFile, timeout)
executor.daemon = True
executor.start()
executor.join()
if executor.timed_out:
out = 'timed out'
else:
outFile.seek(0)
errFile.seek(0)
out = outFile.read()
err = errFile.read()
outFile.close()
errFile.close()
您可以使用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
