Python subprocess模块功能详解：全面介绍与实用示例

Python 的 subprocess 模块用于创建和管理子进程，执行外部命令或脚本。它提供了灵活的方法来启动进程、与进程交互、捕获输出，并处理错误。以下是 subprocess 模块的核心方法及其实际场景的详细说明。

一、subprocess 核心方法

1. subprocess.run()

功能：执行命令并等待其完成，返回 CompletedProcess 对象。
参数：

示例场景：运行 Shell 命令并捕获输出。

import subprocess

result = subprocess.run(
    ["ls", "-l"],         # 推荐使用列表形式（避免 shell 注入）
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True             # 输出为字符串而非字节
)

if result.returncode == 0:
    print("命令成功执行：\n", result.stdout)
else:
    print("命令失败：\n", result.stderr)

2. subprocess.Popen()

功能：更底层的进程创建接口，支持异步操作、实时交互和复杂管道管理。
关键方法：

示例场景：实时读取进程输出并动态处理。

import subprocess

# 启动进程（执行 ping 命令）
proc = subprocess.Popen(
    ["ping", "google.com"],
    stdout=subprocess.PIPE,
    text=True
)

# 逐行读取输出
while True:
    line = proc.stdout.readline()
    if not line and proc.poll() is not None:
        break
    if line:
        print(line.strip())

print("进程退出码：", proc.returncode)

3. 旧版便捷函数

示例场景：快速执行命令并检查结果。

import subprocess

# 检查命令是否成功
try:
    subprocess.check_call(["git", "commit", "-m", "Update code"], check=True)
except subprocess.CalledProcessError as e:
    print("Git 提交失败：", e)

# 获取命令输出
output = subprocess.check_output(["date"], text=True)
print("当前时间：", output)

二、实际应用场景

场景 1：执行 Shell 脚本并传递参数

需求：调用脚本处理文件，传递输入路径和输出路径。

import subprocess

script_path = "process_data.sh"
input_file = "data.csv"
output_file = "result.csv"

result = subprocess.run(
    [script_path, input_file, output_file],
    capture_output=True,
    text=True
)

if result.returncode != 0:
    print("脚本执行失败：", result.stderr)
else:
    print("处理完成：", result.stdout)

场景 2：通过管道传递输入

需求：向进程动态发送输入（如自动化交互式命令行工具）。

import subprocess

# 启动 Python 解释器进程
proc = subprocess.Popen(
    ["python3"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 发送代码并获取输出
commands = """
print("Hello, subprocess!")
x = 1 + 2
print(f"计算结果：{x}")
"""

stdout, stderr = proc.communicate(commands)
print("输出结果：\n", stdout)

场景 3：并行执行多个命令

需求：同时运行多个耗时任务并收集结果。

import subprocess
from concurrent.futures import ThreadPoolExecutor

def run_command(cmd):
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.stdout

commands = [
    ["sleep", "3"],  # 模拟耗时任务
    ["echo", "Hello"],
    ["ls", "/tmp"]
]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(run_command, commands))

for res in results:
    print(res)

场景 4：重定向输出到文件

需求：将命令的输出写入日志文件。

import subprocess

with open("output.log", "w") as f:
    subprocess.run(
        ["python3", "-m", "http.server", "8000"],
        stdout=f,          # 标准输出写入文件
        stderr=subprocess.STDOUT  # 错误输出合并到 stdout
    )

三、关键注意事项

1. 避免 shell=True 的安全风险

subprocess.run(f"rm -rf {user_input}", shell=True)  # 可能被注入恶意命令

subprocess.run(["rm", "-rf", sanitized_user_input])  # 参数列表形式

2. 处理超时

通过 timeout 参数限制命令执行时间：

try:
    subprocess.run(["sleep", "10"], timeout=5)
except subprocess.TimeoutExpired:
    print("命令执行超时！")

3. 跨平台兼容性

四、总结

subprocess 模块的核心方法：

实际场景：

遵循最佳实践（如避免 shell=True、处理超时），可以安全高效地管理子进程。

作者：demonlg0112