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


当前回答

我对底层细节了解不多;但是,考虑到这一点 python 2.6的API提供了等待线程和的能力 终止进程,那么在一个单独的进程中运行该进程呢 的线程吗?

import subprocess, threading

class Command(object):
    def __init__(self, cmd):
        self.cmd = cmd
        self.process = None

    def run(self, timeout):
        def target():
            print 'Thread started'
            self.process = subprocess.Popen(self.cmd, shell=True)
            self.process.communicate()
            print 'Thread finished'

        thread = threading.Thread(target=target)
        thread.start()

        thread.join(timeout)
        if thread.is_alive():
            print 'Terminating process'
            self.process.terminate()
            thread.join()
        print self.process.returncode

command = Command("echo 'Process started'; sleep 2; echo 'Process finished'")
command.run(timeout=3)
command.run(timeout=1)

这段代码在我的机器中的输出是:

Thread started
Process started
Process finished
Thread finished
0
Thread started
Process started
Terminating process
Thread finished
-15

在哪里可以看到,在第一次执行的过程 正确完成(返回代码0),而在第二个 进程被终止(返回代码-15)。

我没有在windows中测试;但是,除了更新示例之外 命令,我想它应该工作,因为我还没有找到 记录任何说明该线程的内容。Join或process.terminate 不支持。

其他回答

从Python 3.5开始,有了一个新的subprocess.run通用命令(意味着取代check_call, check_output…),它也有timeout=参数。

subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None) 执行args描述的命令。等待命令完成,然后返回一个CompletedProcess实例。

它引发一个子流程。超时超时时出现TimeoutExpired异常。

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

#!/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()

我添加了从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

在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

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

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)