Python 基础体系 · 第 47/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python 子进程与信号：参数传递、管道、超时、退出和回收

subprocess 解决的是一个具体问题：父进程如何启动另一个程序，并与它交换标准输入、标准输出、标准错误，等待它结束，再判断它是正常退出还是被信号终止。Python 3.14 中，简单场景优先使用 subprocess.run()；需要分阶段通信、主动发送信号、超时控制或管理进程组时，直接使用 subprocess.Popen。(docs.python.org)

本文以 Linux/POSIX 行为为主，同时指出 Windows 上不能直接照搬的部分。

---

一、先建立模型：子进程不是 Python 函数调用

调用 Python 函数时，调用方和被调用方共享同一个 Python 进程的地址空间；调用子进程时，系统创建了一个独立的进程实体：

父进程
  │
  ├── 创建子进程
  │       ├── 独立 PID
  │       ├── 独立地址空间
  │       ├── 独立文件描述符视图
  │       └── 执行目标程序
  │
  ├── 写入子进程 stdin
  ├── 读取子进程 stdout/stderr
  ├── 向子进程发送信号
  └── 等待并回收子进程

子进程与父进程之间不会自动共享普通 Python 变量。下面的变量只存在于父进程：

command = ["python", "worker.py"]
result = subprocess.run(command)

子进程能看到的内容，必须通过明确的操作传递，例如：

• 命令行参数；
• 环境变量；
• 标准输入；
• 继承或传递的文件描述符；
• 文件、Unix socket、共享内存等外部 IPC；
• 信号，但信号适合表达“事件”或“控制意图”，不适合承载大块数据。

subprocess 本身主要负责“启动程序”和“连接标准流”，并不自动把父进程对象序列化给子进程。(docs.python.org)

---

二、参数传递：列表元素就是参数边界

2.1 推荐使用参数列表

import subprocess

result = subprocess.run(
    [
        "python3",
        "-c",
        "import sys; print(sys.argv[1:])",
        "worker.py",
        "hello world",
        "--count",
        "3",
    ],
    check=True,
    capture_output=True,
    text=True,
)

print(result.stdout, end="")

预期输出：

['hello world', '--count', '3']

这里的参数边界由列表元素决定：

["--name", "hello world"]

表示两个参数：

参数 1：--name
参数 2：hello world

而不是把 hello world 再拆成两个参数。subprocess 文档也建议，通常应传入参数序列，因为模块会处理必要的参数转义和引用；直接传入单个字符串时，含义与 shell 设置有关。(docs.python.org)

2.2 反例：手工拼接命令字符串

import subprocess

name = "report 2026.txt"

# 不推荐
subprocess.run(f"cat {name}", shell=True)

shell 实际可能把它解释为：

cat report 2026.txt

于是 cat 收到两个文件名，而不是一个名为 report 2026.txt 的文件。

更危险的是：

name = "report.txt; rm -rf /tmp/demo"
subprocess.run(f"cat {name}", shell=True)

此时输入数据改变了命令结构。只要命令中包含外部输入，就必须把“数据”与“shell 语法”分开。

正确写法：

import subprocess

subprocess.run(
    ["cat", "report 2026.txt"],
    check=True,
)

此时文件名只是 cat 的一个参数，不会被解释为 shell 命令。shell=True 只有在确实需要 shell 的管道、通配符、变量展开等功能时才应使用，而且必须理解其安全边界。(docs.python.org)

2.3 需要 shell 语法时的边界

import subprocess

subprocess.run(
    "printf '%s\n' *.txt | sort",
    shell=True,
    check=True,
)

这段代码依赖 /bin/sh 解释：

1. *.txt 展开为多个文件名；
2. printf 输出这些文件名；
3. shell 把输出连接到 sort。

如果只是运行一个外部程序，不要因为命令看起来像终端命令就启用 shell：

# 不需要 shell
subprocess.run(["git", "status", "--short"], check=True)

如果程序路径很重要，可以使用绝对路径；需要按 PATH 查找时，可先调用 shutil.which()。再次启动当前 Python 解释器时，推荐使用 sys.executable，而不是硬编码 "python" 或 "python3"。(docs.python.org)

---

三、环境变量和工作目录也是参数的一部分

3.1 环境变量

默认情况下，子进程通常继承父进程的环境。env 不为 None 时，则使用你提供的映射替代默认环境，而不是自动与父环境合并。(docs.python.org)

因此下面的代码可能导致程序找不到可执行文件或动态库：

import subprocess

subprocess.run(
    ["python3", "-c", "print('ok')"],
    env={"MODE": "test"},
    check=True,
)

更常见的写法是复制后修改：

import os
import subprocess

env = os.environ.copy()
env["MODE"] = "test"

subprocess.run(
    ["python3", "-c", "import os; print(os.environ['MODE'])"],
    env=env,
    check=True,
)

输出：

test

这里有两个不同的作用：

• cwd：子进程启动前切换工作目录；
• env：设置子进程看到的环境变量。

subprocess.run(
    ["python3", "-c", "import os; print(os.getcwd())"],
    cwd="/tmp",
    check=True,
)

cwd 不会改变父进程的当前目录，只改变子进程执行时的目录。可执行文件的查找还涉及 PATH、cwd 和平台差异；生产代码若依赖固定程序，应优先使用稳定的绝对路径。(docs.python.org)

---

四、标准输入、标准输出和标准错误：三条独立数据流

每个普通进程通常有三个标准流：

stdin  ：文件描述符 0，进程读取输入
stdout ：文件描述符 1，进程输出正常结果
stderr ：文件描述符 2，进程输出诊断或错误信息

在 Popen 中，它们可以分别设置为：

• None：不重定向，沿用默认行为；
• subprocess.PIPE：创建管道；
• subprocess.DEVNULL：连接到空设备；
• 文件对象或文件描述符；
• stderr=subprocess.STDOUT：把标准错误合并到标准输出。

这些流默认是二进制流；设置 text=True、encoding 或 errors 后，才会以文本模式打开。(docs.python.org)

4.1 一次性执行：run

import subprocess

result = subprocess.run(
    ["python3", "-c", "print('normal'); print('diagnostic', file=__import__('sys').stderr)"],
    capture_output=True,
    text=True,
    check=True,
)

print("returncode =", result.returncode)
print("stdout =", repr(result.stdout))
print("stderr =", repr(result.stderr))

典型输出：

returncode = 0
stdout = 'normal\n'
stderr = 'diagnostic\n'

capture_output=True 等价于同时把 stdout 和 stderr 设为 PIPE，不能再同时显式指定这两个参数。若需要合并两个流，应写成：

result = subprocess.run(
    ["python3", "-c", "import sys; print('out'); print('err', file=sys.stderr)"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    check=True,
)

print(result.stdout)

此时 result.stdout 包含两个流，result.stderr 为 None。(docs.python.org)

4.2 向 stdin 传入数据

import subprocess

result = subprocess.run(
    ["python3", "-c", "import sys; print(sys.stdin.read().upper())"],
    input="hello\nworld\n",
    text=True,
    capture_output=True,
    check=True,
)

print(result.stdout, end="")

输出：

HELLO
WORLD

input 会使内部的 stdin 使用管道，因此不能同时传入 stdin 参数。文本模式下，input 必须是字符串；二进制模式下必须是字节串。(docs.python.org)

---

五、管道的本质：有限容量的内核缓冲区

管道不是无限长的字符串容器，而是一个有容量限制的内核缓冲区：

父进程 ──写──> [ stdin pipe ] ──读──> 子进程
父进程 <─读── [ stdout pipe ] <─写── 子进程
父进程 <─读── [ stderr pipe ] <─写── 子进程

当子进程持续写 stdout，而父进程没有读取时，stdout 管道最终会写满。此后：

子进程 write(stdout) 阻塞
子进程无法继续执行
父进程 wait() 等待子进程结束
双方互相等待

下面的写法存在死锁风险：

import subprocess

proc = subprocess.Popen(
    ["python3", "-c", "print('x' * 10_000_000)"],
    stdout=subprocess.PIPE,
    text=True,
)

proc.wait()                 # 父进程先等待
output = proc.stdout.read() # 可能永远执行不到

子进程输出量足够大时，子进程会因 stdout 管道已满而阻塞；父进程又在 wait()，没有读取 stdout。

正确方式是让 communicate() 同时处理输入、读取 stdout/stderr，并等待进程结束：

import subprocess

proc = subprocess.Popen(
    ["python3", "-c", "print('x' * 10_000_000)"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
)

stdout, stderr = proc.communicate()

print("returncode =", proc.returncode)
print("stdout bytes =", len(stdout))

communicate() 的职责是：

1. 向 stdin 发送数据；
2. 读取 stdout 和 stderr，直到 EOF；
3. 等待子进程结束；
4. 设置 returncode；
5. 返回 (stdout_data, stderr_data)。

但它会把读取到的数据缓存在内存中，因此不适合处理大小无限或无法预估的输出。大输出应改用文件重定向、临时文件，或设计持续消费的读取循环。(docs.python.org)

---

六、run 与 Popen：同步封装和生命周期控制

6.1 run 适合“一次调用，一次结果”

import subprocess

result = subprocess.run(
    ["python3", "-c", "print(42)"],
    capture_output=True,
    text=True,
    timeout=5,
    check=True,
)

print(result.stdout.strip())

run() 会等待程序结束并返回 CompletedProcess。常用结果包括：

• args：实际传入的参数；
• returncode：退出状态；
• stdout：捕获到的标准输出；
• stderr：捕获到的标准错误。

check=True 时，非零退出码会转换为 CalledProcessError 异常。(docs.python.org)

6.2 Popen 适合“创建后分阶段操作”

import subprocess
import time

proc = subprocess.Popen(
    ["sleep", "2"],
)

print("pid =", proc.pid)
print("initial returncode =", proc.poll())  # 通常为 None

time.sleep(0.5)
print("later returncode =", proc.poll())    # 仍可能为 None

code = proc.wait()
print("final returncode =", code)

poll() 不阻塞：子进程仍在运行时返回 None，结束后设置并返回 returncode。wait() 则等待结束并返回退出码。(docs.python.org)

Popen 也支持上下文管理器：

import subprocess

with subprocess.Popen(
    ["python3", "-c", "print('child')"],
    stdout=subprocess.PIPE,
    text=True,
) as proc:
    output = proc.stdout.read()

print(output, end="")

退出 with 块时，标准文件描述符会关闭，进程会被等待。这个机制能减少“忘记关闭管道”或“忘记等待子进程”的生命周期错误。(docs.python.org)

---

七、退出状态：零、非零和负信号编号

7.1 正常退出

在 POSIX 系统上，程序可以调用：

import sys

sys.exit(0)

或者：

raise SystemExit(0)

0 通常表示成功，非零值通常表示失败：

import subprocess

result = subprocess.run(
    ["python3", "-c", "raise SystemExit(7)"],
)

print(result.returncode)

输出：

7

check=True 会把这个非零退出码转换成异常：

import subprocess

try:
    subprocess.run(
        ["python3", "-c", "raise SystemExit(7)"],
        check=True,
    )
except subprocess.CalledProcessError as exc:
    print("exit code:", exc.returncode)

输出：

exit code: 7

7.2 被信号终止

POSIX 下，如果子进程被信号 N 终止，subprocess 中的 returncode 通常表示为 -N。例如：

import signal
import subprocess

proc = subprocess.Popen(["sleep", "60"])
proc.send_signal(signal.SIGTERM)

code = proc.wait()
print(code)

典型输出：

-15

因为 SIGTERM 的编号通常是 15。这里的 -15 不是程序自己调用 exit(-15)，而是 Python 对“由信号终止”的结果编码。官方文档明确说明，负返回值 -N 表示 POSIX 子进程被信号 N 终止。(docs.python.org)

可以这样区分：

import signal
import subprocess

proc = subprocess.Popen(["sleep", "60"])
proc.kill()
code = proc.wait()

if code < 0:
    signum = -code
    print("terminated by", signal.Signals(signum).name)
else:
    print("exited with", code)

在 Linux/POSIX 上，常见结果是：

terminated by SIGKILL

---

八、信号是什么：异步控制事件，而不是普通消息队列

信号是操作系统传递给进程的一种异步事件通知。常见信号包括：

信号	常见含义	默认行为
SIGTERM	请求终止	终止进程
SIGINT	交互式中断，通常来自 Ctrl-C	终止进程
SIGKILL	强制终止	不能捕获、阻塞或忽略
SIGHUP	终端断开或会话变化	通常终止进程
SIGCHLD	子进程状态发生变化	通知父进程

Python 的信号处理有一个容易忽略的时序：底层 C 信号处理器先设置一个标志，Python 解释器稍后在合适的执行点调用 Python 层 handler。因此，长时间运行的纯 C 代码可能不会立即执行 Python handler。Python 信号处理器也总是在主解释器的主线程中执行。(docs.python.org)

例如：

import signal
import time

stop_requested = False

def handle_term(signum, frame):
    global stop_requested
    stop_requested = True
    print("received:", signal.Signals(signum).name)

signal.signal(signal.SIGTERM, handle_term)

while not stop_requested:
    print("working")
    time.sleep(1)

print("cleanup and exit")

handler 中只修改状态，不做复杂清理：

def handle_term(signum, frame):
    global stop_requested
    stop_requested = True

不应在信号处理器中执行长时间阻塞操作、获取可能被其他代码持有的锁或进行复杂 I/O。官方文档特别指出，在信号处理器中使用诸如 threading.Lock 的同步原语可能造成死锁。(docs.python.org)

---

九、向子进程发送信号：send_signal、terminate 和 kill

Popen 提供三个层次不同的方法：

proc.send_signal(signum)
proc.terminate()
proc.kill()

在 POSIX 上：

• send_signal(signal.SIGTERM)：发送指定信号；
• terminate()：发送 SIGTERM；
• kill()：发送 SIGKILL。

SIGTERM 是“请终止”的请求，子进程可以捕获并执行清理；SIGKILL 是强制终止，子进程无法捕获，也没有机会执行 Python 的清理代码。Windows 上，terminate() 和 kill() 的底层语义不同：terminate() 调用 TerminateProcess()，kill() 是其别名；Windows 的 SIGTERM 也与 terminate() 关联。(docs.python.org)

一个分阶段终止流程通常是：

import signal
import subprocess
import time

proc = subprocess.Popen(["python3", "worker.py"])

try:
    proc.wait(timeout=10)
except subprocess.TimeoutExpired:
    print("graceful termination requested")
    proc.send_signal(signal.SIGTERM)

    try:
        proc.wait(timeout=5)
    except subprocess.TimeoutExpired:
        print("force termination")
        proc.kill()
        proc.wait()

print("final returncode:", proc.returncode)

这里的状态转换是：

RUNNING
  │
  ├── 10 秒内退出 ─────────────> EXITED
  │
  └── 超时
        │
        ├── SIGTERM
        │     ├── 5 秒内退出 ─> EXITED
        │     └── 仍未退出
        │
        └── SIGKILL ───────────> KILLED

SIGTERM 和 SIGKILL 不是“两个不同强度的异常”，而是不同的进程终止路径。程序是否能完成临时文件删除、刷新日志、提交事务，取决于它是否获得处理 SIGTERM 的机会。

---

十、超时：等待超时不等于子进程已经停止

10.1 run(timeout=...) 的语义

import subprocess

try:
    subprocess.run(
        ["sleep", "60"],
        timeout=2,
        check=True,
    )
except subprocess.TimeoutExpired as exc:
    print("timed out:", exc.timeout)

对 run() 而言，超时通过内部的 Popen.communicate() 实现；文档说明，超时后子进程会被杀死并等待，随后重新抛出 TimeoutExpired。但创建进程本身在许多平台 API 上不能被中断，因此异常出现的时间可能晚于指定超时时间。(docs.python.org)

10.2 Popen.communicate(timeout=...) 的语义不同

直接使用 Popen 时：

import subprocess

proc = subprocess.Popen(
    ["sleep", "60"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
)

try:
    stdout, stderr = proc.communicate(timeout=2)
except subprocess.TimeoutExpired:
    print("communication timed out")
    proc.kill()
    stdout, stderr = proc.communicate()

print("returncode:", proc.returncode)

这里的关键点是：

1. communicate(timeout=2) 超时；
2. 子进程此时不会自动被杀死；
3. 父进程主动调用 kill()；
4. 再次调用 communicate()，读取残余输出并完成等待；
5. 最终得到 returncode。

超时后不要直接调用 wait() 来代替第二次 communicate()，因为管道中的输出仍需要被消费；官方建议在终止后再次调用 communicate()，而不是调用 wait()。再次通信时不要重新提供 input，否则行为未定义或未来可能报错。(docs.python.org)

10.3 一个可复用的超时清理函数

from __future__ import annotations

import signal
import subprocess
from dataclasses import dataclass


@dataclass
class CommandResult:
    returncode: int
    stdout: str
    stderr: str
    timed_out: bool


def run_with_graceful_timeout(
    args: list[str],
    timeout: float,
    terminate_timeout: float = 3.0,
) -> CommandResult:
    proc = subprocess.Popen(
        args,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )

    try:
        stdout, stderr = proc.communicate(timeout=timeout)
        return CommandResult(
            returncode=proc.returncode,
            stdout=stdout,
            stderr=stderr,
            timed_out=False,
        )

    except subprocess.TimeoutExpired:
        # 第一阶段：请求程序自行清理并退出
        proc.send_signal(signal.SIGTERM)

        try:
            stdout, stderr = proc.communicate(timeout=terminate_timeout)
        except subprocess.TimeoutExpired:
            # 第二阶段：程序不响应，强制终止
            proc.kill()
            stdout, stderr = proc.communicate()

        return CommandResult(
            returncode=proc.returncode,
            stdout=stdout,
            stderr=stderr,
            timed_out=True,
        )

使用：

result = run_with_graceful_timeout(
    ["python3", "-c", "import time; time.sleep(60)"],
    timeout=1,
)

print(result)

该函数的返回状态可能是：

timed_out=True
returncode=-15   # 子进程响应 SIGTERM

或者：

timed_out=True
returncode=-9    # 子进程未响应，最终 SIGKILL

需要注意：这个实现只控制直接子进程。若子进程又启动了孙进程，proc.kill() 默认只作用于直接子进程，孙进程可能继续运行。

---

十一、进程组：为什么只杀 PID 可能不够

考虑下面的结构：

父进程
  └── shell
        └── worker

如果 Python 启动了 shell，而 shell 又启动了真正的 worker，那么：

proc.kill()

通常只杀 proc.pid 对应的进程，不保证整个后代树都被终止。

POSIX 提供进程组和会话来解决这个问题。Popen(start_new_session=True) 会在执行子程序前调用 setsid()，使子进程成为新会话和新进程组的领导者。process_group 则可以在子进程中设置进程组。start_new_session 和 process_group 都是 POSIX 能力；process_group 自 Python 3.11 起提供。(docs.python.org)

11.1 为任务创建独立会话

import os
import signal
import subprocess
import time

proc = subprocess.Popen(
    ["sh", "-c", "sleep 60 & sleep 60"],
    start_new_session=True,
)

try:
    proc.wait(timeout=1)
except subprocess.TimeoutExpired:
    # proc.pid 同时是新进程组的组 ID
    os.killpg(proc.pid, signal.SIGTERM)

    try:
        proc.wait(timeout=3)
    except subprocess.TimeoutExpired:
        os.killpg(proc.pid, signal.SIGKILL)
        proc.wait()

print("returncode:", proc.returncode)

数据流和信号路径如下：

父进程
  │
  ├── Popen 创建新会话
  │       └── 进程组 PGID = proc.pid
  │              ├── shell
  │              ├── sleep
  │              └── sleep
  │
  └── os.killpg(PGID, SIGTERM)
          ├── shell 收到 SIGTERM
          ├── sleep 收到 SIGTERM
          └── sleep 收到 SIGTERM

os.killpg(pgid, sig) 会把信号发送给指定进程组，而不是单个 PID。(docs.python.org)

这里必须确认 proc.pid 确实是新进程组的组 ID。start_new_session=True 的目的就是让子进程调用 setsid()，从而建立独立会话；不能把任意 PID 盲目传给 killpg，否则可能误杀不相关进程组。

11.2 为什么 shell=True 会增加清理复杂度

subprocess.Popen(
    "producer | consumer",
    shell=True,
)

实际启动的通常是：

/bin/sh -c "producer | consumer"

因此父进程直接持有的 PID 可能是 shell，而不是业务程序。管道中还可能存在多个后代进程。若任务必须使用 shell 管道，应该同时设计好进程组、超时和整个任务树的终止策略。

如果可以拆成 Python 管道，应显式连接两个 Popen：

import subprocess

producer = subprocess.Popen(
    ["printf", "b\na\n"],
    stdout=subprocess.PIPE,
)

consumer = subprocess.Popen(
    ["sort"],
    stdin=producer.stdout,
    stdout=subprocess.PIPE,
    text=True,
)

producer.stdout.close()
output, _ = consumer.communicate()

print(output, end="")

producer.stdout.close() 很重要：父进程关闭自己持有的写端或读端副本后，EOF 才能按预期传播。否则某些程序可能一直等待输入结束。

---

十二、退出和回收：进程结束不代表资源已经回收

12.1 僵尸进程是什么

在 POSIX 中，子进程退出后，内核仍会保留一小部分退出信息，例如退出码和资源使用状态，等待父进程读取。父进程通过 wait()、waitpid() 或 Python 的 Popen.wait() 获取这些信息后，子进程才被完全回收。

如果子进程已经结束，而父进程没有等待它，就可能出现僵尸进程：

子进程：已经退出
内核：保留退出状态
父进程：没有 wait
结果：进程表中留下 zombie

Popen.poll() 和 Popen.wait() 都会设置 returncode；使用 Popen 后必须确保存在一条最终等待路径。with Popen(...) 可以帮助自动完成关闭和等待。(docs.python.org)

12.2 waitpid 的底层对应关系

在 POSIX 下，os.waitpid(pid, options) 可以等待指定子进程：

import os
import subprocess

proc = subprocess.Popen(["sh", "-c", "exit 3"])

pid, status = os.waitpid(proc.pid, 0)
code = os.waitstatus_to_exitcode(status)

print(pid == proc.pid)
print(code)

输出：

True
3

waitstatus_to_exitcode() 将底层 wait status 转换为 Python 使用的退出码：正常退出时返回非负退出状态；被信号终止时返回负的信号编号。(docs.python.org)

不过，通常不应混用：

proc.wait()
os.waitpid(proc.pid, 0)

第一次调用已经消费了子进程的等待状态，第二次可能得到 ChildProcessError。除非确实需要使用底层 waitpid 能力，否则让 Popen 管理自己的子进程生命周期更清晰。

---

十三、父进程收到信号时，如何保证子进程不泄漏

一个命令执行器可能在等待子进程时收到 Ctrl-C 或服务停止信号。粗略代码如下：

import subprocess

proc = subprocess.Popen(["long-running-command"])
proc.wait()

如果父进程被 KeyboardInterrupt 打断，子进程未必已经停止。更完整的结构是：

from __future__ import annotations

import os
import signal
import subprocess


def terminate_process_group(proc: subprocess.Popen[str]) -> None:
    if proc.poll() is not None:
        return

    try:
        os.killpg(proc.pid, signal.SIGTERM)
    except ProcessLookupError:
        return

    try:
        proc.wait(timeout=3)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        proc.wait()


proc = subprocess.Popen(
    ["sh", "-c", "trap 'exit 143' TERM; sleep 60"],
    start_new_session=True,
)

try:
    proc.wait()
except KeyboardInterrupt:
    terminate_process_group(proc)
    raise
finally:
    # 即使退出原因不是 KeyboardInterrupt，也要保证回收
    if proc.poll() is None:
        terminate_process_group(proc)

这段代码体现了三个独立责任：

1. 发出终止请求：SIGTERM；
2. 设置强制终止上限：超时后 SIGKILL；
3. 完成回收：最终 wait()。

只发送信号而不等待，会留下未回收的子进程状态；只等待而不终止，可能无限阻塞；只杀直接 PID，则可能留下后代进程。

---

十四、信号处理器与子进程的信号继承

子进程创建时，信号处置可能受到父进程影响。Popen 的 restore_signals=True 默认会把 Python 设置为忽略的部分信号，在执行子程序前恢复为默认处置；文档列出的当前信号包括 SIGPIPE、SIGXFZ 和 SIGXFSZ。(docs.python.org)

这解释了为什么不应依赖 preexec_fn 手工完成所有子进程初始化：

subprocess.Popen(
    command,
    preexec_fn=lambda: os.setsid(),
)

preexec_fn 在带线程的应用中不安全，子进程可能在执行 exec 前因锁状态而死锁。需要新会话时，使用：

subprocess.Popen(
    command,
    start_new_session=True,
)

需要设置进程组时，使用：

subprocess.Popen(
    command,
    process_group=some_pgid,
)

这些专用参数比在 preexec_fn 中调用 setsid() 或 setpgid() 更明确，也避免了相关线程安全问题。(docs.python.org)

---

十五、日志、输出和超时的组合示例

下面的脚本模拟一个持续输出、最终超时的 worker。

15.1 子进程 worker.py

import signal
import sys
import time

stopping = False


def handle_term(signum, frame):
    global stopping
    stopping = True
    print("worker: SIGTERM received", flush=True)


signal.signal(signal.SIGTERM, handle_term)

for index in range(100):
    if stopping:
        print("worker: cleaning up", flush=True)
        sys.exit(143)

    print(f"worker: step={index}", flush=True)
    print(f"worker: diagnostic={index}", file=sys.stderr, flush=True)
    time.sleep(1)

这里使用 flush=True，是为了让每一行尽快写入管道。即使父进程正确读取管道，子进程自己的用户态缓冲也可能导致输出延迟；flush=True 只是示例中的可观测性措施，不是所有程序都能控制。

15.2 父进程 runner.py

from __future__ import annotations

import os
import signal
import subprocess
import sys
import time

proc = subprocess.Popen(
    [sys.executable, "worker.py"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    start_new_session=True,
)

try:
    deadline = time.monotonic() + 3

    while True:
        line = proc.stdout.readline()

        if line:
            print("child>", line, end="")
            continue

        if proc.poll() is not None:
            break

        if time.monotonic() >= deadline:
            raise subprocess.TimeoutExpired(proc.args, 3)

    print("returncode:", proc.returncode)

except subprocess.TimeoutExpired:
    print("timeout: terminate process group")

    os.killpg(proc.pid, signal.SIGTERM)

    try:
        proc.wait(timeout=2)
    except subprocess.TimeoutExpired:
        print("timeout: kill process group")
        os.killpg(proc.pid, signal.SIGKILL)
        proc.wait()

    print("final returncode:", proc.returncode)

finally:
    if proc.stdout is not None:
        proc.stdout.close()

运行：

python3 runner.py

可能输出：

child> worker: step=0
child> worker: diagnostic=0
child> worker: step=1
child> worker: diagnostic=1
child> worker: step=2
child> worker: diagnostic=2
timeout: terminate process group
final returncode: -15

这个示例没有使用 communicate()，因为它需要在进程运行期间逐行观察输出。代价是父进程必须自己处理：

• stdout 是否出现 EOF；
• 子进程是否已退出；
• 超时截止时间；
• 进程组终止；
• 最终文件描述符关闭。

如果不需要实时输出，应优先使用 communicate()，因为它更容易正确地同时消费 stdout 和 stderr。

---

十六、常见错误与诊断路径

错误一：把参数列表当成 shell 命令

错误：

subprocess.run(["echo hello world"])

这表示执行一个名为 echo hello world 的程序，而不是执行 echo 并传入两个参数。

正确：

subprocess.run(["echo", "hello", "world"], check=True)

错误二：只读取 stdout，不读取 stderr

proc = subprocess.Popen(
    command,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
)

stdout = proc.stdout.read()
proc.wait()

如果 stderr 写满，子进程可能阻塞在 stderr，导致 stdout 永远读不到 EOF。应使用：

stdout, stderr = proc.communicate()

或者把 stderr 合并到 stdout：

proc = subprocess.Popen(
    command,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
)

错误三：超时后只捕获异常，不清理

错误：

try:
    proc.communicate(timeout=5)
except subprocess.TimeoutExpired:
    print("timeout")

此时 Popen.communicate() 超时不会自动杀死子进程。必须发送终止信号，并再次完成通信或等待。(docs.python.org)

错误四：kill() 后不 wait()

错误：

proc.kill()
return

kill() 只是向子进程发送强制终止信号，并不等于父进程已经读取退出状态。正确流程仍然是：

proc.kill()
proc.wait()

如果有管道，则更稳妥：

proc.kill()
stdout, stderr = proc.communicate()

错误五：误以为 SIGTERM 一定立即终止

子进程可以捕获 SIGTERM，也可以在 handler 中延迟退出、忽略信号，或者因阻塞在不可中断的系统操作中而暂时不退出。因此生产超时控制通常需要：

SIGTERM
  │
  └── 等待宽限期
          │
          └── SIGKILL

错误六：使用 shell=True 后只杀 shell PID

shell 可能启动多个后代进程。任务超时后只调用：

proc.kill()

可能只结束 shell，真正的 worker 仍然运行。POSIX 下应考虑 start_new_session=True 与 os.killpg() 的组合。(docs.python.org)

---

十七、如何选择 API

可以按控制需求选择：

只运行命令，等待结果
    └── subprocess.run()

需要传入 stdin，并捕获 stdout/stderr
    └── subprocess.run(input=..., capture_output=True)

需要实时读取输出
    └── Popen + 读取循环

需要在运行中发送信号
    └── Popen + send_signal()/terminate()/kill()

需要超时后分级终止
    └── Popen + communicate(timeout) + 终止 + 再 communicate()

需要清理整个任务树
    └── POSIX: start_new_session=True + os.killpg()

需要大规模或无限输出
    └── 避免 communicate() 将全部输出放入内存

最终需要区分四个不同动作：

1. 启动：Popen(...)；
2. 通信：communicate() 或主动读写管道；
3. 终止：send_signal()、terminate()、kill()；
4. 回收：wait()、communicate() 或上下文管理器退出。

只有四个阶段都闭合，子进程生命周期才完整：

创建
  ↓
运行与通信
  ↓
正常退出 / 信号终止 / 超时
  ↓
读取退出状态
  ↓
关闭管道并回收

subprocess 的难点不在于“执行一条命令”，而在于同时维护参数边界、管道背压、信号语义、超时状态和进程回收这几条独立的控制路径。任何一条路径没有完成，都可能表现为命令挂起、输出丢失、子进程泄漏、僵尸进程或残留后台任务。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python 随机数与 secrets：伪随机、采样、种子和密码学边界
• 下一篇：Python 日志工程：Logger、Handler、结构化字段、上下文和轮转
• 延伸：Python 多进程：启动方式、IPC、共享内存、Pool 与回收
• 延伸：Python 自动化脚本：文件、命令、网络、重试、幂等和审计

官方资料

• Python 3.14 Standard Library
• Python HOWTOs

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。