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

我有一个问题，我想终止一个多线程子进程，如果它花费的时间超过给定的超时长度。我想在Popen()中设置一个超时，但它不起作用。然后，我意识到Popen().wait()等于call()，所以我有了在.wait(timeout=xxx)方法中设置超时的想法，这最终工作了。因此，我是这样解决的:

import os
import sys
import signal
import subprocess
from multiprocessing import Pool

cores_for_parallelization = 4
timeout_time = 15  # seconds

def main():
    jobs = [...YOUR_JOB_LIST...]
    with Pool(cores_for_parallelization) as p:
        p.map(run_parallel_jobs, jobs)

def run_parallel_jobs(args):
    # Define the arguments including the paths
    initial_terminal_command = 'C:\\Python34\\python.exe'  # Python executable
    function_to_start = 'C:\\temp\\xyz.py'  # The multithreading script
    final_list = [initial_terminal_command, function_to_start]
    final_list.extend(args)

    # Start the subprocess and determine the process PID
    subp = subprocess.Popen(final_list)  # starts the process
    pid = subp.pid

    # Wait until the return code returns from the function by considering the timeout. 
    # If not, terminate the process.
    try:
        returncode = subp.wait(timeout=timeout_time)  # should be zero if accomplished
    except subprocess.TimeoutExpired:
        # Distinguish between Linux and Windows and terminate the process if 
        # the timeout has been expired
        if sys.platform == 'linux2':
            os.kill(pid, signal.SIGTERM)
        elif sys.platform == 'win32':
            subp.terminate()

if __name__ == '__main__':
    main()

仅适用于Linux，但如果有人想使用subprocess.getstatusoutput()，其中timeout参数不可用，您可以在命令的开头使用内置的Linux timeout，即:

import subprocess

timeout = 25 # seconds
cmd = f"timeout --preserve-status --foreground {timeout} ping duckgo.com"
exit_c, out = subprocess.getstatusoutput(cmd)

if (exit_c == 0):
    print("success")
else:
    print("Error: ", out)

超时参数:

——preserve-status:保留退出状态 ——前台:在前台运行 25:超时时间，单位为秒

一旦你理解了在*unix中运行机器的整个过程，你将很容易找到更简单的解决方案:

考虑这个简单的例子，如何使用select.select()(现在几乎在*nix上随处可见)创建可超时的communication()冰毒。这也可以用epoll/poll/kqueue来编写，但select.select()变体可能是一个很好的例子。select.select()的主要限制(速度和1024 max fds)不适用于您的任务。

它在*nix下工作，不创建线程，不使用信号，可以从任何线程启动(不仅仅是主线程)，并且足够快，可以从我机器上的stdout读取250mb/s的数据(i5 2.3ghz)。

在通信结束时连接stdout/stderr时出现问题。如果你有大量的程序输出，这可能会导致大量的内存使用。但是您可以多次调用communication()，超时时间较小。

class Popen(subprocess.Popen):
    def communicate(self, input=None, timeout=None):
        if timeout is None:
            return subprocess.Popen.communicate(self, input)

        if self.stdin:
            # Flush stdio buffer, this might block if user
            # has been writing to .stdin in an uncontrolled
            # fashion.
            self.stdin.flush()
            if not input:
                self.stdin.close()

        read_set, write_set = [], []
        stdout = stderr = None

        if self.stdin and input:
            write_set.append(self.stdin)
        if self.stdout:
            read_set.append(self.stdout)
            stdout = []
        if self.stderr:
            read_set.append(self.stderr)
            stderr = []

        input_offset = 0
        deadline = time.time() + timeout

        while read_set or write_set:
            try:
                rlist, wlist, xlist = select.select(read_set, write_set, [], max(0, deadline - time.time()))
            except select.error as ex:
                if ex.args[0] == errno.EINTR:
                    continue
                raise

            if not (rlist or wlist):
                # Just break if timeout
                # Since we do not close stdout/stderr/stdin, we can call
                # communicate() several times reading data by smaller pieces.
                break

            if self.stdin in wlist:
                chunk = input[input_offset:input_offset + subprocess._PIPE_BUF]
                try:
                    bytes_written = os.write(self.stdin.fileno(), chunk)
                except OSError as ex:
                    if ex.errno == errno.EPIPE:
                        self.stdin.close()
                        write_set.remove(self.stdin)
                    else:
                        raise
                else:
                    input_offset += bytes_written
                    if input_offset >= len(input):
                        self.stdin.close()
                        write_set.remove(self.stdin)

            # Read stdout / stderr by 1024 bytes
            for fn, tgt in (
                (self.stdout, stdout),
                (self.stderr, stderr),
            ):
                if fn in rlist:
                    data = os.read(fn.fileno(), 1024)
                    if data == '':
                        fn.close()
                        read_set.remove(fn)
                    tgt.append(data)

        if stdout is not None:
            stdout = ''.join(stdout)
        if stderr is not None:
            stderr = ''.join(stderr)

        return (stdout, stderr)

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

另一种选择是写入临时文件以防止标准输出阻塞，而不需要使用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()