下面是运行任意命令返回其标准输出数据的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程序中实现超时的最简单方法是什么?


当前回答

有时需要处理(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:
.......

其他回答

预先设置Linux命令超时并不是一个糟糕的解决方法,它对我来说是有效的。

cmd = "timeout 20 "+ cmd
subprocess.Popen(cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(output, err) = p.communicate()

在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后端端口实现。

我添加了从jcollado线程到我的Python模块easyprocess的解决方案。

安装:

pip install easyprocess

例子:

from easyprocess import Proc

# shell is not supported!
stdout=Proc('ping localhost').call(timeout=1.5).stdout
print stdout

虽然我还没有广泛地研究它,但我在ActiveState中发现的这个装饰器似乎对这类事情非常有用。伴随着subprocess.Popen(…, close_fds=True),至少我已经准备好在Python中编写shell脚本。

您可以使用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