python 执行外部命令

脚本语言
发布时间 :
字数:2.1k 阅读 :
  1. 1 subprocess
    1. 1.1 核心函数 run
      1. 1.1.1 用法
      2. 1.1.2 举例说明
      3. 1.1.3 返回值对象
    2. 1.2 高级函数 Popen
      1. 1.2.1 用法
      2. 1.2.2 举例说明
      3. 1.2.3 返回值对象
      4. 1.2.4 返回 pid
    3. 1.3 应用示例
  2. 2 os.system

在 Linux 系统中，我们经常会编写 shell 脚本来执行任务，或者调用系统的一些命令，这对于 python 来讲，属于执行外部命令，python 标准库里面提供了一些接口，这里做一个介绍。

1 subprocess

subprocess 是 Python 标准库中用于 创建和管理子进程 的核心模块。它允许你从 Python 程序中启动外部命令（如 shell 脚本、可执行程序、系统工具等），并与它们进行交互（传递参数、读取输出、捕获错误、控制超时等）。

应用场景：

  • 执行 shell 脚本或系统命令（bash, ffmpeg, rsync 等）
  • 调用其他语言编写的程序（如 R、MATLAB 脚本）
  • 自动化运维（备份、部署、监控）
  • 与 Docker、Git、AWS CLI 等工具集成
  • 并行任务调度（配合 concurrent.futures）

subprocess 主要提供了两个接口：run()和Popen()

特性 subprocess.run() subprocess.Popen()
引入版本 Python 3.5（推荐新代码使用） Python 2.4（旧但功能强大）
阻塞行为 同步阻塞：调用后立即等待子进程结束 异步非阻塞：启动后立即返回，不等待
使用复杂度 简单、高层、一行搞定 复杂、底层、需手动管理
适用场景 一次性执行命令，获取结果 需要精细控制、实时交互、长时间运行
返回值 CompletedProcess 对象 Popen 对象（可后续操作）

1.1 核心函数 run

1.1.1 用法

subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None,
               capture_output=False, timeout=None, check=False, text=None,
               encoding=None, errors=None, env=None, cwd=None, ...)

同步阻塞：调用后立即等待子进程结束

常用参数说明：

参数 作用
args 要执行的命令，推荐使用列表形式：["ls", "-l"]（更安全）；也可用字符串 "ls -l"（但需 shell=True）
capture_output 若为 True，自动设置 stdout=PIPE, stderr=PIPE，便于获取输出
text=True（或 universal_newlines=True） 返回字符串而非 bytes
timeout 超时时间（秒），超时则抛出 TimeoutExpired
check=True 如果子进程返回非零退出码，自动抛出 CalledProcessError
cwd 指定子进程的工作目录
env 设置环境变量（字典）

1.1.2 举例说明

import subprocess

result1 = subprocess.run(["ls", "-l"],
    capture_output=True,
    text=True
)
print(" 返回码:", result1.returncode)
print(" 标准输出:\n", result1.stdout)
print(" 标准错误:\n", result1.stderr)

result2 = subprocess.run(["sort"],
    input="banana\napple\ncherry",
    text=True,
    capture_output=True
)
print(result2.stdout)

超时与异常处理：

try:
    result = subprocess.run(["sleep", "10"],
        timeout=2,
        check=True
    )
except subprocess.TimeoutExpired:
    print(" 命令执行超时！")
except subprocess.CalledProcessError as e:
    print(f" 命令失败，返回码: {e.returncode}")

1.1.3 返回值对象

subprocess.run() 返回一个 CompletedProcess 对象，包含：

  • .args：原始命令
  • .returncode：退出状态码（0 通常表示成功）
  • .stdout：标准输出（bytes 或 str）
  • .stderr：标准错误
  • .check_returncode()：手动检查是否失败（类似 check=True）

1.2 高级函数 Popen

subprocess.Popen 是 Python subprocess 模块中最底层、最灵活的子进程管理接口。

它允许你 启动一个子进程后立即返回，非阻塞式，并在后续对其执行精细控制（如读写标准流、检查状态、发送信号、等待结束等）。

适用于需要 实时交互、长时间运行或并发管理多个进程 的场景。

1.2.1 用法

subprocess.Popen(args, bufsize=-1, stdin=None, stdout=None, stderr=None, shell=False,
    cwd=None, env=None, text=None, encoding=None, errors=None, ...
)

常用参数说明：

参数 说明
args 命令参数，推荐列表形式：["cmd", "arg1", "arg2"]；若用字符串需设 shell=True
stdin, stdout, stderr 控制标准流： - None：继承父进程（默认） - subprocess.PIPE：创建管道，可读写 - subprocess.DEVNULL：丢弃 - 文件对象：重定向到文件
shell 是否通过 shell 执行（如 bash）。慎用，有安全风险
cwd 子进程工作目录
env 环境变量字典（若为 None 则继承父进程）
text（或 universal_newlines） 若为 True，以字符串而非 bytes 读写流
bufsize 缓冲策略： - 0：无缓冲（仅二进制模式） - 1：行缓冲（文本模式） - -1：系统默认（推荐）

1.2.2 举例说明

import subprocess
import sys

proc = subprocess.Popen(["ping", "-c", "5", "google.com"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    bufsize=1  # 行缓冲
)

# 实时打印每一行
for line in proc.stdout:
    print(line.strip())
    sys.stdout.flush()  # 确保立即输出

proc.wait()  # 务必调用

print("Ping 结束，返回码:", proc.returncode)

不要丢弃 Popen 对象而不等待：会导致僵尸进程（zombie process）。

始终确保调用 .wait()、.communicate() 或检查 .poll()，以回收子进程资源。

1.2.3 返回值对象

subprocess.Popen() 返回一个 Popen 对象（类型为 subprocess.Popen），它是对底层操作系统子进程的 封装句柄，允许你与该子进程进行交互、查询状态、读写数据、发送信号等。

1. 主要属性：

属性 类型 说明
.args list 或 str 启动时传入的命令参数
.pid int 子进程的操作系统进程 ID（PID）
.returncode int 或 None - None：进程仍在运行 - 整数（如 0, 1）：进程已结束，表示退出码

2. 主要方法

方法 说明
.poll() 检查子进程是否结束，不阻塞。 - 返回 None：仍在运行 - 返回整数：退出码
.wait(timeout=None) 阻塞等待 子进程结束，返回退出码。 超时抛 TimeoutExpired
.communicate(input=None, timeout=None) 一次性 发送输入 + 读取输出 + 等待结束。 避免死锁（比直接读 .stdout.read() 安全）
.send_signal(signal) 发送信号（如 signal.SIGTERM）
.terminate() 发送 SIGTERM（优雅终止）
.kill() 发送 SIGKILL（强制杀死）

1.2.4 返回 pid

调用方式 .pid对应的进程
Popen(["./script.sh"]) ./script.sh（前提是它有 #!/bin/bash 且可执行）
Popen(["/bin/bash", "script.sh"]) /bin/bash 进程
Popen("script.sh", shell=True) 系统默认 shell（如 /bin/sh）进程

无论哪种方式，.pid 始终是 Popen 直接创建的那个“父进程”的 PID，而不是脚本内部启动的某个子命令的 PID。

1.3 应用示例

假设有一个任务如下：用 python 提交一个 shell 脚本，这个 shell 脚本执行一个外部命令 cad -f test.rule。这个外部命令是一个耗时的命令。应该怎么做？

import subprocess
import concurrent.futures
import time
import signal
import os

WORK_DIR = "/path/to/your/workdir"  # cad 所需的工作目录

def run_cad_task(task_id):

    try:
        # 启动子进程
        proc = subprocess.Popen(["./run_cad.sh"],
            cwd=WORK_DIR,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            preexec_fn=os.setsid  # 创建新进程组（便于后续杀死整个进程树）
        )

        # 等待完成或超时
        try:
            stdout, stderr = proc.communicate(timeout=3600*24)
            returncode = proc.returncode
        except subprocess.TimeoutExpired:
            # 超时：杀死整个进程组
            try:
                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)  # 发送终止信号 signal.SIGTERM
                proc.wait(timeout=10)
            except:
                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)  # 发送强杀信号 signal.SIGKILL
                proc.wait()
            return (task_id, False, f" 任务超时 ")

        # 检查结果
        if returncode == 0:
            return (task_id, True, " 成功完成 ")
        else:
            return (task_id, False, f" 失败，退出码: {returncode}\nstderr: {stderr[-500:]}")  # 截断长错误

    except Exception as e:
        return (task_id, False, f" 异常: {str(e)}")


def main():

    task_ids = [f"task_{i}" for i in range(10)]  # 示例：10 个任务

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        # 提交所有任务
        future_to_task = {
            executor.submit(run_cad_task, tid): tid
            for tid in task_ids
        }

        # 按完成顺序处理结果
        for future in concurrent.futures.as_completed(future_to_task):
            task_id, success, msg = future.result()
            if success:
                print(f"[{task_id}] {msg}")
            else:
                print(f"[{task_id}] {msg}")

if __name__ == "__main__":
    main()

为什么用 ThreadPoolExecutor？

  • subprocess.Popen 启动的是 独立 OS 进程，Python 线程只是在“等待”。
  • 线程池开销小，适合管理大量外部命令。
  • ProcessPoolExecutor 会 fork Python 进程，造成不必要的内存复制。

2 os.system

os.system() 是 Python 标准库 os 模块中的一个 简单 的函数，用于在子 shell 中执行操作系统命令。

自 Python 2.4（2004 年）引入 subprocess 后，os.system() 就已 被标记为遗留接口。

import os

os.system("command")  # 将字符串 command 传递给操作系统的默认 shell 执行

命令的 输出会直接打印到终端，无法在 Python 程序中捕获或处理。

命令的执行过程

  1. 创建一个子进程；
  2. 子进程启动系统 shell；
  3. shell 解析并执行你传入的命令字符串；
  4. 父进程（Python）阻塞等待子进程结束；
  5. 返回子进程的退出状态。
欢迎各位看官及技术大佬前来交流指导呀，可以邮件至 jqiange@yeah.net

©2016-2021 Jqiange

Built with Hexo and 3-hexo theme