Python subprocess 详解

核心概念 Python subprocess 模块用于创建子进程并与其交互。

1. 核心定位

subprocess 是 Python 官方标准库，唯一推荐用于创建子进程、调用外部系统命令/可执行程序的模块，彻底淘汰老旧废弃接口：os.system()、os.popen()、commands。

适用场景：

Linux 运维执行 df -h、ss、systemctl、curl 等 shell 命令；
调用第三方二进制程序（ffmpeg、mysql、kubectl、docker）；
捕获命令标准输出、错误输出、返回码；
给子进程传入标准输入、控制超时、工作目录、自定义环境变量；
异步后台执行、实时流式读取日志（tail -f）、管道串联多条命令。底层原理：
Python 父进程通过操作系统 fork/exec 创建独立子进程，父子进程拥有独立内存空间，通过管道 pipe 实现 stdin/stdout/stderr 数据互通。

2. API 层级划分（从高层简单到底层灵活）

2.1. 高层封装（Python3.5+ 首选：`subprocess.run()`）

日常 90% 场景只用它，封装了等待、捕获输出、异常校验逻辑。

2.2. 老旧高层（仅兼容旧代码，新项目禁用）

call() / check_call() / check_output()

2.3. 底层原生核心（复杂场景必用）

subprocess.Popen()，所有高层接口底层都是封装 Popen；

支持异步不阻塞、实时逐行输出、长后台进程、多管道串联。

三、subprocess.run() 完整参数详解

3. 基础语法


subprocess.run(

    args,

    stdin=None,

    stdout=None,

    stderr=None,

    shell=False,

    cwd=None,

    env=None,

    timeout=None,

    check=False,

    text=False,

    universal_newlines=False,

    capture_output=False,

    close_fds=True

)

4. 逐个核心参数拆解

4.1. args：命令传入两种模式（安全重中之重）

4.1.1. 模式1：`shell=False` 默认，列表传参【推荐、防命令注入】

命令拆分为字符串列表，不会交给 shell 解析，特殊字符自动转义，不存在注入漏洞，接收用户输入必须用这种写法。


# 正确，执行 ls -l /tmp

res = subprocess.run(["ls", "-l", "/tmp"], capture_output=True, text=True)

错误写法（shell=False 下传字符串会直接报文件不存在）：


subprocess.run("ls -l /tmp") # FileNotFoundError

4.1.2. 模式2：`shell=True` 字符串整条命令，慎用

完整命令作为字符串传入，由系统 shell（Linux /bin/sh、Windows cmd）解析；

支持管道 |、通配符 *、重定向 >、分号多条命令；

致命风险：存在命令注入漏洞，绝对不能拼接外部用户输入


# 管道、通配符只有 shell=True 才支持

res = subprocess.run("df -h | grep /dev/vda", shell=True, capture_output=True, text=True)

注入危险示例（禁止）：


user_input = ";rm -rf /data/*;"

# 高危！恶意输入会执行删除命令

subprocess.run(f"echo {user_input}", shell=True)

# 安全替代方案

subprocess.run(["echo", user_input])

4.2. stdout / stderr 三种取值

None（默认）：输出直接打印到终端，Python 代码无法捕获；
subprocess.PIPE：将输出捕获到内存，可通过返回对象读取；
文件对象：重定向输出写入本地文件；简写：capture_output=True 等价 stdout=PIPE, stderr=PIPE。

4.3. text / universal_newlines

text=False（默认）：输出是 bytes 字节流，需要手动 .decode("utf-8")；
text=True：自动解码为字符串，运维脚本必开，省去转码、解决中文乱码；
universal_newlines 是 text 旧别名，功能完全一致。

4.4. check=True 自动校验返回码

命令执行失败（returncode != 0）直接抛出 subprocess.CalledProcessError，不用手动判断返回码：


# 不存在的命令直接抛异常

subprocess.run(["xxx_command_xxx"], check=True, capture_output=True, text=True)

4.5. timeout 超时控制

子进程运行超过指定秒数，操作系统强制杀死进程，抛出 subprocess.TimeoutExpired，防止脚本卡死阻塞。


# 最多运行2秒，超时直接终止

subprocess.run(["sleep", "10"], timeout=2, capture_output=True)

4.6. cwd 指定执行工作目录

切换命令执行目录，无需手动调用 os.chdir()，执行完自动复原父进程目录：


# 在 /data/log 下执行 pwd

subprocess.run(["pwd"], cwd="/data/log", capture_output=True, text=True)

4.7. env 自定义环境变量

子进程默认完全继承父进程 Python 的环境变量；传入字典可以覆盖/新增环境变量：


import os

new_env = os.environ.copy()

new_env["DEBUG"] = "1"

subprocess.run(["echo", "$DEBUG"], shell=True, env=new_env)

5. run() 返回值：CompletedProcess 对象

执行完成后返回实例，核心属性：


res = subprocess.run(...)

res.args        # 执行的完整命令

res.returncode  # 进程退出码，0=正常，非0=执行异常

res.stdout      # 标准输出字符串/bytes

res.stderr      # 标准错误输出

6. run 通用运维封装函数（生产直接复用）


import subprocess

def exec_cmd(cmd_list, timeout=10):

    """

    执行系统命令，返回 (退出码, stdout, stderr)

    :param cmd_list: shell=False 列表命令

    """

    try:

        res = subprocess.run(

            cmd_list,

            capture_output=True,

            text=True,

            timeout=timeout

        )

        return res.returncode, res.stdout.strip(), res.stderr.strip()

    except subprocess.TimeoutExpired as e:

        return -1, "", f"命令执行超时：{e}"

# 使用示例

code, out, err = exec_cmd(["df", "-h"])

if code == 0:

    print("磁盘信息：\n", out)

else:

    print("执行失败：", err)

四、底层核心：Popen 类（复杂场景专用）

run() 内部只是对 Popen 的封装，以下场景必须直接使用 Popen：

不需要阻塞等待，后台异步启动进程；
实时流式逐行读取输出（tail -f、打包日志、编译输出）；
多条命令管道串联 cmd1 | cmd2；
向子进程动态输入 stdin 数据；
长时间后台运行，手动 terminate/kill 杀死进程。

7. Popen 基础用法


import subprocess

# 创建子进程：非阻塞，代码立刻向下执行，不会等待命令结束

p = subprocess.Popen(

    ["ping", "127.0.0.1", "-c", "4"],

    stdout=subprocess.PIPE,

    stderr=subprocess.PIPE,

    text=True

)

# 阻塞等待进程结束，获取退出码

ret_code = p.wait()

# 一次性读取全部输出（阻塞）

out, err = p.communicate()

print(out)

8. communicate() 方法详解

阻塞等待子进程运行完毕；
一次性读取全部 stdout、stderr；
支持传入 input 字符串，作为子进程标准输入；


# 向cat命令传入文本

p = subprocess.Popen(["cat"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)

out, err = p.communicate(input="测试标准输入内容")

print(out) # 输出：测试标准输入内容

⚠️ 短板：必须等进程结束才能拿到输出，不适合实时打印日志。

9. 实时流式逐行读取（运维高频：实时打印日志）

适用于 tail -f、程序编译、长时间输出场景，边输出边打印，不用等进程结束：


def stream_cmd(cmd):

    p = subprocess.Popen(

        cmd,

        stdout=subprocess.PIPE,

        text=True

    )

    # 循环逐行读取管道输出

    while True:

        line = p.stdout.readline()

        if not line:

            break

        print(line.strip())

    p.wait()

# 实时跟踪系统日志

stream_cmd(["tail", "-f", "/var/log/messages"])

10. 多管道串联（等价 shell `ls | grep txt`）

不开启 shell=True，纯 Python 实现管道，安全无注入风险：


# 第一层 ls

p1 = subprocess.Popen(["ls"], stdout=subprocess.PIPE, text=True)

# 第二层 grep，stdin 绑定p1的输出管道

p2 = subprocess.Popen(["grep", "txt"], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)

# 关闭p1管道句柄，避免管道缓冲区死锁

p1.stdout.close()

out, _ = p2.communicate()

print(out)

11. 手动终止后台进程


# 后台睡眠100秒

p = subprocess.Popen(["sleep", "100"])

# 温和终止 SIGTERM

p.terminate()

# 强制杀死 SIGKILL（Linux）

# p.kill()

# 回收进程，防止产生僵尸进程

p.wait()

12. 进程状态判断


p.is_alive() # 返回 True/False，判断进程是否仍在运行

p.pid        # 获取操作系统真实进程ID，start后才有值

五、三大常见异常类

FileNotFoundError
args 中的可执行程序不存在、路径错误；shell=False 场景最容易触发。
subprocess.CalledProcessError
check=True / check_call / check_output 时，进程退出码非0，代表命令执行报错；异常对象自带 stdout/stderr。
subprocess.TimeoutExpired
设置 timeout 参数，命令运行超时被强制杀死；可通过异常对象读取已输出内容。

六、高频踩坑点（运维脚本必避）

13. shell=True 命令注入漏洞

只要输入内容来自用户、接口、外部文件，一律禁止 shell=True，使用列表传参自动转义。

14. 管道缓冲区满导致死锁

子进程输出大量内容填满管道缓冲区会阻塞进程；

解决方案：

使用 communicate() 一次性读取全部输出；
stderr 重定向到 stdout：stderr=subprocess.STDOUT；
实时逐行读取管道。

15. 中文乱码

未开启 text=True 获取 bytes 字节，打印出现乱码；开启 text=True 自动按系统编码解码。


# 乱码

res = subprocess.run(["echo", "中文"], stdout=subprocess.PIPE)

print(res.stdout)

# 正常

res = subprocess.run(["echo", "中文"], stdout=subprocess.PIPE, text=True)

16. 僵尸进程

使用 Popen 创建后台进程后，不调用 wait() / communicate() 回收，进程结束后会长期残留僵尸进程占用PID；

解决方案：所有 Popen 进程最终必须调用 wait()。

17. Windows / Linux 跨平台差异

Linux shell=True 调用 /bin/sh；Windows 调用 cmd.exe；
Windows 程序后缀 .exe/.bat，路径分隔符 \ 需要转义 \\；
部分系统命令参数格式不一致（ping、tasklist / ps）。

18. atexit 资源泄露

长期后台脚本（监控、AIOps采集程序）频繁创建 Popen 不回收，大量句柄耗尽触发 too many open files。

七、老旧废弃API（仅了解，新项目禁止使用）

subprocess.call()：只返回退出码，无法捕获输出
subprocess.check_call()：执行失败抛异常，不返回输出
subprocess.check_output()：仅捕获stdout，stderr直接打印终端替代方案：统一使用带 capture_output=True, check=True 的 subprocess.run()。

八、运维场景完整实战示例

19. 示例1：批量执行kubectl命令，捕获输出


code, out, err = exec_cmd(["kubectl", "get", "pod", "-n", "default"])

if code == 0:

    print(out)

20. 示例2：超时+自动异常捕获（监控巡检脚本标准模板）


try:

    subprocess.run(

        ["systemctl", "status", "nginx"],

        timeout=3,

        check=True,

        capture_output=True,

        text=True

    )

except subprocess.TimeoutExpired as e:

    print("查询nginx状态超时", e.stderr)

except subprocess.CalledProcessError as e:

    print("nginx服务异常", e.stderr)

21. 示例3：向程序传入标准输入


# 执行一段Python代码

res = subprocess.run(

    ["python3", "-"],

    input="print('通过stdin传入代码')",

    capture_output=True,

    text=True

)

print(res.stdout)

九、面试核心速记总结

subprocess 替代 os.system/popen，分高层 run()、底层 Popen；
shell=False 列表传参安全防注入；shell=True 支持管道通配符，禁止拼接外部输入；
stdout/stderr=PIPE 捕获输出，text=True 自动转字符串避免bytes乱码；
run() 阻塞等待进程结束；Popen 异步创建，communicate/wait 回收进程；
Popen 适合实时流式输出、多管道、后台长进程；
三大异常：命令不存在、返回码非0、执行超时；
避坑要点：外部输入禁用shell、Popen必须wait防僵尸进程、管道缓冲区满会死锁；
运维标准封装：统一封装run函数，带回退出码、标准输出、错误信息。

关联文档

Python Linux 系统运维详解

psutil 系统监控详解

Python subprocess 详解

1. 核心定位

2. API 层级划分（从高层简单到底层灵活）

2.1. 高层封装（Python3.5+ 首选：subprocess.run()）

2.2. 老旧高层（仅兼容旧代码，新项目禁用）

2.3. 底层原生核心（复杂场景必用）

三、subprocess.run() 完整参数详解

3. 基础语法

4. 逐个核心参数拆解

4.1. args：命令传入两种模式（安全重中之重）

4.1.1. 模式1：shell=False 默认，列表传参【推荐、防命令注入】

4.1.2. 模式2：shell=True 字符串整条命令，慎用

4.2. stdout / stderr 三种取值

4.3. text / universal_newlines

4.4. check=True 自动校验返回码

4.5. timeout 超时控制

4.6. cwd 指定执行工作目录

4.7. env 自定义环境变量

5. run() 返回值：CompletedProcess 对象

6. run 通用运维封装函数（生产直接复用）

四、底层核心：Popen 类（复杂场景专用）

7. Popen 基础用法

8. communicate() 方法详解

9. 实时流式逐行读取（运维高频：实时打印日志）

10. 多管道串联（等价 shell ls | grep txt）

11. 手动终止后台进程

12. 进程状态判断

五、三大常见异常类

六、高频踩坑点（运维脚本必避）

13. shell=True 命令注入漏洞

14. 管道缓冲区满导致死锁

15. 中文乱码

16. 僵尸进程

17. Windows / Linux 跨平台差异

18. atexit 资源泄露

七、老旧废弃API（仅了解，新项目禁止使用）

八、运维场景完整实战示例

19. 示例1：批量执行kubectl命令，捕获输出

20. 示例2：超时+自动异常捕获（监控巡检脚本标准模板）

21. 示例3：向程序传入标准输入

九、面试核心速记总结

目录

2.1. 高层封装（Python3.5+ 首选：`subprocess.run()`）

4.1.1. 模式1：`shell=False` 默认，列表传参【推荐、防命令注入】

4.1.2. 模式2：`shell=True` 字符串整条命令，慎用

10. 多管道串联（等价 shell `ls | grep txt`）