下面是运行任意命令返回其标准输出数据的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程序中实现超时的最简单方法是什么?
我已经修改了sussudio的答案。现在函数返回:(returncode, stdout, stderr, timeout) - stdout和stderr被解码为utf-8字符串
def kill_proc(proc, timeout):
timeout["value"] = True
proc.kill()
def run(cmd, timeout_sec):
proc = subprocess.Popen(shlex.split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
timeout = {"value": False}
timer = Timer(timeout_sec, kill_proc, [proc, timeout])
timer.start()
stdout, stderr = proc.communicate()
timer.cancel()
return proc.returncode, stdout.decode("utf-8"), stderr.decode("utf-8"), timeout["value"]
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
另一种选择是写入临时文件以防止标准输出阻塞,而不需要使用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()
我已经实现了我能从其中一些收集到的东西。这适用于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()
在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后端端口实现。