对子进程的运行时间做限制，当超时时强制终止

发表于 阅读次数： 本文字数： 950 阅读时长 ≈ 3 分钟

终止运行超时的进程

subprocess 模块主要用于创建子进程，并连接它们的输入、输出和错误管道，获取它们的返回状态，在某种特殊情况下，我们需要对子进程的运行时间做限制，当超时时强制终止并获取输出。
subprocess 模块首先推荐使用的是它的 run 方法，更高级的用法是使用 Popen。

Popen 方法语法格式如下：

1
subprocess.Popen(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=False, startupinfo=None, creationflags=0,restore_signals=True, start_new_session=False, pass_fds=(),*, encoding=None, errors=None)

常用参数：

  • args：表示要执行的命令，可以是字符串或者序列类型（如：list，元组）
  • bufsize：缓冲区大小。当创建标准流的管道对象时使用，默认-1。
    0：不使用缓冲区
    1：表示行缓冲，仅当universal_newlines=True时可用，也就是文本模式
    正数：表示缓冲区字节大小
    负数：表示使用系统默认的缓冲区大小。
  • stdin、stdout 和 stderr：子进程的标准输入、输出和错误。其值可以是 subprocess.PIPE、subprocess.DEVNULL、一个已经存在的文件描述符、已经打开的文件对象或者 None。subprocess.PIPE 表示为子进程创建新的管道。subprocess.DEVNULL 表示使用 os.devnull。默认使用的是 None，表示什么都不做。另外，stderr 可以合并到 stdout 里一起输出。
  • timeout：设置命令超时时间。如果命令执行时间超时，子进程将被杀死，并弹出 TimeoutExpired 异常。
  • check：如果该参数设置为 True，并且进程退出状态码不是 0，则弹 出 CalledProcessError 异常。
  • encoding: 如果指定了该参数，则 stdin、stdout 和 stderr 可以接收字符串数据，并以该编码方式编码。否则只接收 bytes 类型的数据。
  • shell：如果该参数为 True，将通过操作系统的 shell 执行指定的命令(bash或cmd)。
  • cwd：用于设置子进程的当前目录。
  • env：用于指定子进程的环境变量。如果 env = None，子进程的环境变量将从父进程中继承。

直接用 timeout 参数为何不能实现?

当运行的命令自身超过这个时间限制时，不会触发 TimeoutExpired 异常。这是因为 subprocess.Popen 的 timeout 参数是设置子进程的超时时间，而不是执行的命令的超时时间，所以当执行的命令自身超过此时间限制时无法通过 timeout 参数来实现。

解决方案

Timer（定时器）是 Thread 的派生类，用于在指定时间后调用一个方法。

参数介绍：

  • interval — 定时器间隔，间隔多少秒之后启动定时器任务(单位：秒)；
  • function — 线程函数；
  • args — 线程参数，可以传递元组类型数据，默认为空(缺省参数)；
  • kwargs — 线程参数，可以传递字典类型数据，默认为空(缺省参数)；
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import subprocess
from threading import Timer


def kill_command(process):
    """终止命令的函数"""
    # poll(): 检查进程是否终止，如果终止返回 returncode，否则返回 None
    if process.poll() is None:
        process.kill() # 或 process.terminate()

def deduplicate(stdout):
    """
    去掉输出的重复行
    """
    stdout_lines = stdout.splitlines()
    uniq_lines = []
    for line in stdout_lines:
        if line not in uniq_lines:
            uniq_lines.append(line)
    deduplicated_stdout = "\n".join(uniq_lines)

    return deduplicated_stdout

def execute_command(command,timeout):
    """
    用子进程执行命令，捕获日志输出
    :param command: 待执行的命令
    :param timeout: 超时时间
    :return: 返回command和错误信息
    """
    proc = subprocess.Popen(command,stdout=subprocess.PIPE,stderr=subprocess.PIPE,encoding="utf-8")
    # 设置定时器当达到 timeout 时终止这个子进程
    timer = Timer(timeout, kill_command, [proc])
    timer.start()
    # 获取输出与退出码
    stdout = proc.communicate()[0]
    return_code = proc.returncode
    log = deduplicate(stdout)
    # 当运行时间低于 timeout 时间则取消定时器
    timer.cancel()
    # 值为 false 时超时，true 未超时
    not_timeout = timer.is_alive()
    # 返回超时的信息
    if return_code != 0:
        return command,log

execute_command("ls -l & sleep 10",5)