首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置: 程式碼之家 >> 編程語言 >  >> 更多程式語言 >> 更多程式設計知識

Python subprocess 模組完整指南:進程管理、管道通訊、錯誤處理與最佳實務

     編輯:更多程式設計知識

subprocess 是 Python 3 標準庫中管理子進程的核心模組,取代舊式 os.system、os.popen、commands,提供統一、安全、功能完整的進程建立、通訊、控制介面。本文系統整理核心 API、常用模式、管道串聯、即時互動、錯誤處理與生產級最佳實務。

核心 API 比較

函式/類別用途阻塞回傳適用場景
subprocess.run()執行命令、等待完成是CompletedProcessPython 3.5+ 首選、簡單同步執行
subprocess.Popen()建立進程、非阻塞控制否Popen 物件複雜互動、管道、長時間運行、背景進程
subprocess.call()執行、回傳退出碼是int (exit code)舊版相容、只需退出碼
subprocess.check_output()執行、回傳 stdout是bytes/str需捕獲輸出、失敗拋例外
subprocess.check_call()執行、非零拋例外是0需嚴格檢查成功

基礎用法:subprocess.run() (Python 3.5+ 推薦)


import subprocess

# 1. 基本執行 (列表形式,避免 shell 注入)
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
print(result.returncode)  # 0
print(result.stdout)      # 標準輸出
print(result.stderr)      # 標準錯誤

# 2. 檢查失敗自動拋例外
try:
    subprocess.run(["false"], check=True)
except subprocess.CalledProcessError as e:
    print(f"命令失敗: {e.returncode}")

# 3. 輸入資料
result = subprocess.run(["cat"], input="hello\nworld", capture_output=True, text=True)
print(result.stdout)  # hello\nworld

# 4. 環境變數、工作目錄、超時
result = subprocess.run(
    ["./myapp"],
    env={**os.environ, "CUSTOM_VAR": "value"},
    cwd="/opt/app",
    timeout=30,
    capture_output=True,
    text=True
)

進階控制:Popen 非阻塞進程管理


import subprocess
import time

# 建立進程
proc = subprocess.Popen(
    ["ping", "google.com"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 非阻塞讀取 (輪詢)
while True:
    # 檢查是否結束
    ret = proc.poll()
    if ret is not None:
        print(f"進程結束, 退出碼: {ret}")
        break
    
    # 讀取一行輸出
    line = proc.stdout.readline()
    if line:
        print(f"[stdout] {line.strip()}")
    
    # 避免 CPU 佔用
    time.sleep(0.1)

# 等待結束並取得剩餘輸出
stdout, stderr = proc.communicate()

管道串聯:進程間資料流


# 等同 shell: cat file.txt | grep "error" | wc -l
cat = subprocess.Popen(["cat", "file.txt"], stdout=subprocess.PIPE)
grep = subprocess.Popen(["grep", "error"], stdin=cat.stdout, stdout=subprocess.PIPE)
wc = subprocess.Popen(["wc", "-l"], stdin=grep.stdout, stdout=subprocess.PIPE)

# 關鍵:關閉上游 stdout,讓下游收到 EOF
cat.stdout.close()
grep.stdout.close()

result = wc.communicate()[0]
print(f"錯誤行數: {result.decode().strip()}")

即時互動:stdin/stdout 即時讀寫


proc = subprocess.Popen(
    ["python3", "-c", "import sys; [print(f'echo: {line.strip()}') for line in sys.stdin]"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    bufsize=1  # 行緩衝
)

# 發送命令
for cmd in ["hello", "world", "test"]:
    proc.stdin.write(cmd + "\n")
    proc.stdin.flush()  # 強制刷新
    
    # 即時讀取回應
    response = proc.stdout.readline()
    print(f"收到: {response.strip()}")

proc.stdin.close()
proc.wait()

輸出重導向:檔案、變數、丟棄


# 1. 導向檔案
with open("output.log", "w") as f:
    subprocess.run(["ls", "-l"], stdout=f, stderr=subprocess.STDOUT)

# 2. 導向變數
result = subprocess.run(["ls"], capture_output=True, text=True)
output = result.stdout

# 3. 丟棄輸出 (靜默執行)
subprocess.run(["noisy_cmd"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

# 4. 合併 stderr 到 stdout
result = subprocess.run(["cmd"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

Shell=True 的風險與正確用法

模式範例風險建議
shell=False (預設)run(["ls", "-l"])無注入風險、參數明確分隔首選,安全
shell=Truerun("ls -l", shell=True)命令注入、參數解析歧義、效能較差僅需 shell 功能(管道、通配符、變數展開)時使用

# 危險範例:使用者輸入注入
user_input = "test; rm -rf /"
subprocess.run(f"echo {user_input}", shell=True)  # 執行了 rm -rf / !

# 安全範例:列表形式,參數不被解析
subprocess.run(["echo", user_input])  # 只輸出 "test; rm -rf /"

錯誤處理最佳實務


import subprocess
import sys

def run_command(cmd, **kwargs):
    """統一命令執行封裝"""
    defaults = {
        'capture_output': True,
        'text': True,
        'timeout': 60,
    }
    defaults.update(kwargs)
    
    try:
        result = subprocess.run(cmd, **defaults)
        if result.returncode != 0:
            raise subprocess.CalledProcessError(
                result.returncode, cmd, result.stdout, result.stderr
            )
        return result.stdout
    except subprocess.TimeoutExpired:
        raise RuntimeError(f"命令超時: {cmd}")
    except FileNotFoundError:
        raise RuntimeError(f"命令不存在: {cmd[0]}")
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"命令失敗 [{e.returncode}]: {e.stderr}")

# 使用
try:
    output = run_command(["docker", "build", "-t", "myapp", "."])
    print("建構成功")
except RuntimeError as e:
    print(f"建構失敗: {e}")
    sys.exit(1)

跨平台考量

議題WindowsUnix/Linux/macOS
路徑分隔符\/
可執行檔副檔名.exe, .bat, .cmd無 (權限位)
Shellcmd.exe / PowerShell/bin/sh, /bin/bash
信號處理有限 (CTRL_C_EVENT, CTRL_BREAK_EVENT)完整 (SIGTERM, SIGKILL, SIGINT...)
進程群組Job ObjectsProcess Groups (os.setsid)

# 跨平台尋找可執行檔
import shutil
python_exe = shutil.which("python3") or shutil.which("python")
if not python_exe:
    raise RuntimeError("Python 未安裝")

# 跨平台啟動獨立進程
if sys.platform == "win32":
    proc = subprocess.Popen(cmd, creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
else:
    proc = subprocess.Popen(cmd, start_new_session=True)

常見陷阱與避坑

陷阱症狀解決
deadlock (管道滿)大量輸出時卡住用 communicate()、執行緒讀取、或增加 bufsize
殭屍進程子進程結束但父進程未 wait呼叫 wait()/communicate()、或設定 SIGCHLD 處理
編碼錯誤輸出含非 ASCII 字元報錯指定 text=True, encoding='utf-8', errors='replace'
環境變數未繼承子進程找不到 PATH 程式傳遞 env={**os.environ, ...}
超時未殺進程樹timeout 只殺父進程用 process group (start_new_session) 或 psutil 清理

總結

subprocess 是 Python 進程管理的標準解決方案:簡單同步用 run()、複雜控制用 Popen()。核心最佳實務:列表形式命令 (shell=False)、capture_output/text 參數、通過 check=True 檢查失敗、communicate() 避免死鎖、start_new_session 隔離進程群組。掌握管道串聯、即時互動、輸出重導向、錯誤處理、跨平台差異,能構建健壯的系統整合、自動化部署、CLI 工具封裝等生產級應用。

  1. 上一篇文章:
  2. 下一篇文章:
Copyright © 程式碼之家 All Rights Reserved