Python中创建和管理子进程:subprocess



一、subprocess.run —— 同步执行(带注释)

import subprocess

result = subprocess.run(
    ["ls", "-l"],              # args:命令及参数(推荐列表形式,避免 shell 注入)
    stdin=None,                # stdin:标准输入,None 表示继承父进程
    stdout=subprocess.PIPE,    # stdout:捕获标准输出
    stderr=subprocess.PIPE,    # stderr:捕获错误输出
    capture_output=False,      # True 等价于 stdout+stderr=PIPE
    shell=False,               # 是否通过 shell 执行(强烈建议 False)
    cwd=None,                  # 指定子进程工作目录
    timeout=10,                # 超时时间(秒),超时抛 TimeoutExpired
    check=False,               # True:返回码非 0 抛异常
    text=True,                 # True:stdout/stderr 为 str(否则是 bytes)
    encoding="utf-8",          # 文本编码(text=True 时有效)
    errors="ignore",           # 编码错误处理方式
    env=None                   # 子进程环境变量(dict)
)

# 返回对象:CompletedProcess
print(result.args)        # 实际执行的命令
print(result.returncode)  # 返回码(0 表示成功)
print(result.stdout)      # 标准输出(str 或 bytes)
print(result.stderr)      # 错误输出


二、⭐ subprocess.Popen —— 异步 / 可控 / 有 PID(重点)

1️⃣ 创建子进程(逐参数注释)

import subprocess

p = subprocess.Popen(
    ["ping", "baidu.com", "-c", "5"],  # args:命令列表(推荐)
    bufsize=1,                         # 缓冲区大小(1 表示行缓冲)
    executable=None,                   # 替换默认可执行文件
    stdin=subprocess.PIPE,             # 标准输入管道
    stdout=subprocess.PIPE,            # 标准输出管道
    stderr=subprocess.PIPE,            # 错误输出管道
    preexec_fn=None,                   # Unix:子进程 exec 前执行函数
    close_fds=True,                    # 关闭多余文件描述符
    shell=False,                       # 是否使用 shell
    cwd=None,                          # 工作目录
    env=None,                          # 环境变量
    text=True,                         # 等价 universal_newlines=True
    encoding="utf-8",                  # 输出编码
    errors="ignore",                   # 编码错误处理
    start_new_session=False            # True:创建新会话(进程组)
)


2️⃣ Popen 对象的核心属性(带注释)

p.pid          # 子进程 PID(shell=True 时是 shell 的 PID)
p.stdin        # 写入子进程 stdin
p.stdout       # 读取子进程 stdout
p.stderr       # 读取子进程 stderr
p.returncode   # 进程退出码,None 表示仍在运行


3️⃣ Popen 的核心方法(带注释)

p.poll()
# None   → 进程仍在运行
# int    → 进程已结束(返回码)

p.wait(timeout=None)
# 阻塞等待进程结束
# 返回 returncode
# timeout 超时抛 TimeoutExpired

stdout, stderr = p.communicate(input=None, timeout=None)
# 向 stdin 写入 input(一次性)
# 读取所有 stdout / stderr
# 会阻塞直到进程结束(不适合实时输出)

p.terminate()
# 发送 SIGTERM(优雅终止)

p.kill()
# 发送 SIGKILL(强制杀死)


4️⃣ ⭐ 实时读取 stdout(最常用 executor 模式)

p = subprocess.Popen(
    ["ping", "baidu.com", "-c", "5"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 实时读取标准输出
for line in p.stdout:
    print("OUT:", line.rstrip())

# 等待进程结束
code = p.wait()

print("return code:", code)


三、stdin / stdout / stderr 的取值说明(注释版)

subprocess.PIPE      # 建立管道(Python 可读写)
subprocess.DEVNULL   # 丢弃输出
None                 # 继承父进程(直接打印到终端)
文件对象              # 重定向到文件

示例:

with open("out.log", "w") as f:
    subprocess.run(
        ["ls"],
        stdout=f
    )


四、shell=True 的注释说明(⚠️ 很重要)

p = subprocess.Popen(
    "ls | grep py",  # shell 才支持管道
    shell=True
)

⚠️ 实际进程结构:

Python
 └── /bin/sh   ← p.pid
     └── ls / grep

风险:

  • PID 不是目标命令
  • 容易命令注入
  • 不利于进程管理

👉 executor / agent 强烈不推荐



五、进程组管理(带注释,生产常用)

import os
import signal
import subprocess

p = subprocess.Popen(
    ["sleep", "100"],
    preexec_fn=os.setsid  # 创建新进程组(Unix)
)

pgid = os.getpgid(p.pid)

# 杀掉整个进程组(包括孙进程)
os.killpg(pgid, signal.SIGTERM)


六、异常类型(带注释)

try:
    subprocess.run(
        ["false"],
        check=True
    )
except subprocess.CalledProcessError as e:
    print(e.returncode)   # 返回码
    print(e.cmd)          # 命令
except subprocess.TimeoutExpired as e:
    print("timeout:", e.timeout)


七、老接口(注释说明,不推荐)

subprocess.call(args)
# 同 run(args).returncode

subprocess.check_call(args)
# 非 0 抛异常,无输出

subprocess.check_output(args)
# 只返回 stdout,非 0 抛异常

👉 现在统一用 run()



八、终极总结(注释版口诀 🧠)

run      → 简单、同步、无 PID
Popen    → 异步、可控、有 PID(executor 首选)
PIPE     → Python 读取
shell    → 能不用就不用


文章标签:

评论(0)