下面是运行任意命令返回其标准输出数据的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程序中实现超时的最简单方法是什么?
当前回答
没想到居然没人提到超时
Timeout 5 ping -c 3 somehost
显然，这并不适用于每个用例，但如果您处理的是一个简单的脚本，那么这是很难克服的。
mac用户也可以通过homebrew在coreutils中使用gtimeout。
其他回答
下面是Alex Martelli的解决方案，它是一个具有适当进程终止的模块。其他方法不起作用，因为它们没有使用proc. communication()。所以如果你有一个产生大量输出的进程，它会填满它的输出缓冲区，然后阻塞，直到你从中读取一些东西。
from os import kill
from signal import alarm, signal, SIGALRM, SIGKILL
from subprocess import PIPE, Popen
def run(args, cwd = None, shell = False, kill_tree = True, timeout = -1, env = None):
'''
Run a command with a timeout after which it will be forcibly
killed.
'''
class Alarm(Exception):
pass
def alarm_handler(signum, frame):
raise Alarm
p = Popen(args, shell = shell, cwd = cwd, stdout = PIPE, stderr = PIPE, env = env)
if timeout != -1:
signal(SIGALRM, alarm_handler)
alarm(timeout)
try:
stdout, stderr = p.communicate()
if timeout != -1:
alarm(0)
except Alarm:
pids = [p.pid]
if kill_tree:
pids.extend(get_process_children(p.pid))
for pid in pids:
# process might have died before getting to this line
# so wrap to avoid OSError: no such process
try:
kill(pid, SIGKILL)
except OSError:
pass
return -9, '', ''
return p.returncode, stdout, stderr
def get_process_children(pid):
p = Popen('ps --no-headers -o pid --ppid %d' % pid, shell = True,
stdout = PIPE, stderr = PIPE)
stdout, stderr = p.communicate()
return [int(p) for p in stdout.split()]
if __name__ == '__main__':
print run('find /', shell = True, timeout = 3)
print run('find', shell = True)
不幸的是，我的雇主在公开源代码方面有非常严格的规定，所以我不能提供实际的代码。但对我来说，最好的解决方案是创建一个子类重写Popen.wait()来轮询，而不是无限期地等待。__init__接受超时参数。一旦你这样做，所有其他Popen方法(调用等待)将按预期工作，包括通信。
在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
虽然我还没有广泛地研究它，但我在ActiveState中发现的这个装饰器似乎对这类事情非常有用。伴随着subprocess.Popen(…， close_fds=True)，至少我已经准备好在Python中编写shell脚本。
有一个想法是子类化Popen类并用一些简单的方法装饰器来扩展它。我们叫它ExpirablePopen吧。
from logging import error
from subprocess import Popen
from threading import Event
from threading import Thread
class ExpirablePopen(Popen):
def __init__(self, *args, **kwargs):
self.timeout = kwargs.pop('timeout', 0)
self.timer = None
self.done = Event()
Popen.__init__(self, *args, **kwargs)
def __tkill(self):
timeout = self.timeout
if not self.done.wait(timeout):
error('Terminating process {} by timeout of {} secs.'.format(self.pid, timeout))
self.kill()
def expirable(func):
def wrapper(self, *args, **kwargs):
# zero timeout means call of parent method
if self.timeout == 0:
return func(self, *args, **kwargs)
# if timer is None, need to start it
if self.timer is None:
self.timer = thr = Thread(target=self.__tkill)
thr.daemon = True
thr.start()
result = func(self, *args, **kwargs)
self.done.set()
return result
return wrapper
wait = expirable(Popen.wait)
communicate = expirable(Popen.communicate)
if __name__ == '__main__':
from subprocess import PIPE
print ExpirablePopen('ssh -T git@bitbucket.org', stdout=PIPE, timeout=1).communicate()
