Python subprocess详解
subprocess 模块是 Python 官方推荐用于创建和管理子进程的标准库，它统一并替代了 os.system、os.popen 等旧式方法。它能让你安全、灵活地执行外部命令，并获取其输入输出。
🚀 核心函数：run() 与 Popen()
subprocess 提供了两个主要接口，你可以按需选用：
-
subprocess.run()(推荐): 这是 Python 3.5 之后的首选方式。它用于执行一个命令并等待其完成，返回一个包含执行结果的CompletedProcess对象。适合大多数一次性调用场景。 -
subprocess.Popen()(高级): 这是更底层的接口，run()也是基于它实现的-。Popen对象代表一个正在运行的子进程，允许你在进程启动后与之持续交互、检查状态等。适合需要复杂交互、实时流式处理或非阻塞式管理的场景。
run() 常用参数详解:
subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, capture_output=False, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, env=None, universal_newlines=None, **other_popen_kwargs)
运行被 args 描述的指令。等待指令完成，然后返回一个 CompletedProcess 实例。
| 参数 | 说明 | 示例 |
|---|---|---|
args |
要执行的命令，推荐使用列表形式（如 ["ls", "-l"]），更安全。 |
subprocess.run(["echo", "Hello"]) |
capture_output |
若为 True，则捕获子进程的 stdout 和 stderr。 |
subprocess.run(..., capture_output=True) |
text |
若为 True，则输入输出以文本 (字符串) 形式处理，否则为 bytes 字节流。 |
subprocess.run(..., text=True) |
check |
若为 True，当子进程返回非零退出码时抛出 CalledProcessError 异常。 |
subprocess.run(..., check=True) |
timeout |
设置超时时间（秒），超时后会抛出 TimeoutExpired 异常。 |
subprocess.run(..., timeout=10) |
input |
向子进程的标准输入发送数据。 | subprocess.run(..., input="y\n") |
env |
为子进程设置自定义的环境变量字典。 | subprocess.run(..., env={"PATH": "/custom/bin"}) |
cwd |
设置子进程的当前工作目录。 | subprocess.run(..., cwd="/path/to/dir") |
🔐 安全第一：警惕 shell=True
这是使用 subprocess 时最需要警惕的风险点。
-
风险: 当
shell=True时，命令会通过系统的 shell（如/bin/sh或cmd.exe）执行。这意味着命令字符串中的特殊字符（如;,|,&,$等）会被 shell 解析，极易引发命令注入攻击。 -
错误示例:
# 极度危险！切勿使用！ user_input = "file.txt; rm -rf /" subprocess.run(f"cat {user_input}", shell=True) # 可能执行删除操作 -
正确做法: 尽量避免使用
shell=True。大多数情况下，直接传递参数列表是安全且有效的。# 安全做法 filename = "file with space.txt" subprocess.run(["cat", filename]) # 参数列表，安全可靠 -
何时使用: 仅当命令必须依赖 shell 特性（如通配符
*.txt、管道|、环境变量$HOME）且命令本身是静态的、不包含任何用户输入时，才考虑使用。
🛠️ 核心技巧与进阶用法
1. 捕获与处理输出
-
分别捕获: 使用
capture_output=True可以分别获取stdout和stderr。 -
合并输出: 如果你想将标准错误合并到标准输出中，可以设置
stderr=subprocess.STDOUT。 -
实时输出: 对于耗时命令，
run()会等待进程结束才返回。如需实时获取输出，需使用Popen逐行读取stdout。
import subprocess
# 启动进程
process = subprocess.Popen(
["ping", "-c", "5", "google.com"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1 # 行缓冲
)
# 实时读取输出（逐行）
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
print(output.strip())
# 获取剩余输出和返回码
stdout, stderr = process.communicate()2. 输入与管道
-
提供输入: 通过
input参数可以向子进程的标准输入发送数据。subprocess.run(["grep", "error"], input="line1\nerror line", text=True) -
构建管道: 如需实现 Shell 中的管道（如
cmd1 | cmd2），可以手动创建两个Popen对象，并将第一个的stdout连接到第二个的stdin。
使用 subprocess 实现 Shell 管道（cmd1 | cmd2）的核心原理是：将第一个进程的 stdout 连接到第二个进程的 stdin。
在 Python 中，最标准且安全的方法是手动创建两个 Popen 对象，并利用 subprocess.PIPE 来连接它们。下面我将从最基础的实现开始，逐步深入到链式管道和错误处理。
核心实现：连接两个命令
下面以 echo "hello world" | grep "world" 为例：
import subprocess
# 1. 启动第一个进程，将其 stdout 重定向到管道 (PIPE)
p1 = subprocess.Popen(
["echo", "hello world"],
stdout=subprocess.PIPE,
text=True # 以文本模式处理
)
# 2. 启动第二个进程，将其 stdin 设置为 p1 的 stdout
p2 = subprocess.Popen(
["grep", "world"],
stdin=p1.stdout, # 关键连接点：从 p1 读取输出
stdout=subprocess.PIPE,
text=True
)
# ⚠️ 极其关键的一步：在父进程中关闭 p1 的 stdout
p1.stdout.close()
# 3. 获取最终结果（从 p2 读取）
output, error = p2.communicate()
print(f"输出: {output}") # 输出: hello world
print(f"p1 返回码: {p1.wait()}") # 通常为 0
print(f"p2 返回码: {p2.returncode}") # 0为什么必须关闭 p1.stdout.close()？
-
p1.stdout在父进程（你的 Python 脚本）中也持有管道的读取端。 -
如果不关闭它，当
p2处理完数据退出后，p1可能仍在尝试写入数据。 -
更严重的是，如果
p2提前退出（比如只匹配前几行就退出），p1会因为管道缓冲区写满而永久挂起，因为管道的另一端（父进程持有的那个）始终没有关闭。 -
关闭后，当
p2退出时，p1会收到SIGPIPE信号并自动终止，或者正常写完所有数据后退出。
进阶实现：链式管道（cmd1 | cmd2 | cmd3）
假设我们要实现 cat file.txt | grep "error" | wc -l，只需按顺序串联即可。核心原则是：始终关闭上一个进程的 stdout。
import subprocess
# 构建管道链
p1 = subprocess.Popen(["cat", "file.txt"], stdout=subprocess.PIPE, text=True)
p2 = subprocess.Popen(["grep", "error"], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)
p1.stdout.close() # p1 的输出已交给 p2，父进程不再需要
p3 = subprocess.Popen(["wc", "-l"], stdin=p2.stdout, stdout=subprocess.PIPE, text=True)
p2.stdout.close() # p2 的输出已交给 p3，父进程不再需要
# 获取最终输出
result, _ = p3.communicate()
print(f"错误行数: {result.strip()}")
# 确保所有进程都被回收（如果它们还没退出）
p1.wait()
p2.wait()
p3.wait()封装为通用函数（便于复用）：
import subprocess
def safe_pipeline(commands):
procs = []
for i, cmd in enumerate(commands):
if i == 0:
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True)
else:
p = subprocess.Popen(cmd, stdin=procs[-1].stdout, stdout=subprocess.PIPE, text=True)
procs[-1].stdout.close()
procs.append(p)
output, _ = procs[-1].communicate()
# 等待所有进程结束并检查返回码
for p in procs:
p.wait()
if p.returncode != 0:
# 抛出异常或进行错误处理，注意管道中 grep 无匹配时返回 1 是正常的
raise subprocess.CalledProcessError(p.returncode, p.args)
return output
try:
# 故意让 grep 匹配不到（会返回 1）
result = safe_pipeline([
["echo", "abc"],
["grep", "xyz"]
])
except subprocess.CalledProcessError as e:
print(f"管道中的命令失败: {e}")补充技巧：同时获取 stderr
如果你想分别获取每个命令的错误信息，可以为每个 Popen 设置 stderr=subprocess.PIPE，然后通过 p.communicate() 或后续读取 p.stderr 来获取。
3. 超时控制
为 run() 或 Popen.communicate() 设置 timeout 参数，可以防止子进程永久挂起-。
try:
subprocess.run(["sleep", "10"], timeout=5)
except subprocess.TimeoutExpired:
print("命令执行超时！")4. 环境变量与工作目录
-
环境变量: 使用
env参数可以为子进程设置独立的变量，而不是继承父进程的。 -
工作目录: 使用
cwd参数可以指定子进程的执行路径。
5. 错误处理
设置 check=True 可以让 run() 在命令失败时抛出异常，这比手动检查 returncode 更简洁。
try:
subprocess.run(["false"], check=True)
except subprocess.CalledProcessError as e:
print(f"命令失败，返回码: {e.returncode}")⚠️ 避坑指南
-
避免字符串拼接: 构建命令参数时，始终使用列表而非字符串拼接，以防止注入和参数解析错误。
-
注意文本与字节: 使用
text=True处理文本，使用bytes处理二进制数据。 -
跨平台兼容性: Windows 和 Unix 的命令、路径风格不同，编写脚本时需注意。例如，Windows 上可执行文件通常有
.exe后缀。 -
僵尸进程: 使用
Popen时，务必调用proc.wait()或proc.communicate()来回收子进程资源，避免产生僵尸进程-。 -
PIPE死锁: 如果使用PIPE且数据量很大，必须及时读取，否则可能导致死锁。communicate()方法可以安全地处理这个问题-。
import subprocess
# 启动进程
process = subprocess.Popen(
["ping", "-c", "5", "google.com"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1 # 行缓冲
)
# 实时读取输出（逐行）
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
print(output.strip())
# 获取剩余输出和返回码
stdout, stderr = process.communicate()-
Popen作为上下文管理器: 从 Python 3.2 开始，Popen支持上下文管理器，可以确保资源被正确清理-。with subprocess.Popen(["cmd"], stdout=subprocess.PIPE) as proc: print(proc.communicate()[0])
💎 总结
使用 subprocess 模块，记住以下三点：
-
优先使用
subprocess.run()，它足以应对 90% 的需求。 -
默认使用参数列表，除非绝对必要，否则永远不要开启
shell=True。 -
善用
capture_output、text和check等参数，编写更健壮的代码。
参考资料：
更多推荐
8
0- 0
已为社区贡献12条内容
所有评论(0)