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


当前回答

只是想写些简单点的东西。

#!/usr/bin/python

from subprocess import Popen, PIPE
import datetime
import time 

popen = Popen(["/bin/sleep", "10"]);
pid = popen.pid
sttime = time.time();
waittime =  3

print "Start time %s"%(sttime)

while True:
    popen.poll();
    time.sleep(1)
    rcode = popen.returncode
    now = time.time();
    if [ rcode is None ]  and  [ now > (sttime + waittime) ] :
        print "Killing it now"
        popen.kill()

其他回答

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

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

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

timeout现在由子进程模块中的call()和communication()支持(从Python3.3开始):

import subprocess

subprocess.call("command", timeout=20, shell=True)

这将调用该命令并引发异常

subprocess.TimeoutExpired

如果命令在20秒后还没有完成。

然后你可以处理异常来继续你的代码,就像这样:

try:
    subprocess.call("command", timeout=20, shell=True)
except subprocess.TimeoutExpired:
    # insert code here

希望这能有所帮助。

这是我的解决方案,我使用线程和事件:

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)