subprocess 是 Python 3 標準庫取代 os.system、os.popen、commands 的統一進程管理模組。核心類 Popen 提供精細的標準流控制 (stdin/stdout/stderr)、管道鏈接、非阻塞 I/O、超時處理、信號傳遞。本文深入解析 Popen 進程間通訊 (IPC) 實戰模式、關鍵參數、死鎖風險與最佳實踐。
Popen 核心參數
import subprocess
proc = subprocess.Popen(
args, # 命令列表 ['cmd', 'arg1'] 或字串 (shell=True)
stdin=None, # None/PIPE/檔案物件/檔案描述符
stdout=None, # 同上
stderr=None, # 同上
shell=False, # True: 經由 shell 解析 (支援管道、重導向、通配符)
cwd=None, # 工作目錄
env=None, # 環境變數字典 (None 繼承父進程)
universal_newlines=False, # True: 文字模式 (str)，False: 二進制模式 (bytes)
encoding=None, # 文字模式編碼 (Python 3.6+)
errors=None, # 編碼錯誤處理
bufsize=-1, # 緩衝區大小 (-1 系統預設, 0 無緩衝, >0 指定)
start_new_session=False, # True: 建立新會話 (成為進程組長)
pass_fds=(), # 保留的檔案描述符 (Unix)
)
import subprocess
# 讀取命令輸出 (阻塞直到結束)
proc = subprocess.Popen(['ipconfig', '/all'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = proc.communicate() # 等待結束，返回 (stdout, stderr)
print(f'返回碼: {proc.returncode}')
print(f'標準輸出:\n{stdout.decode("cp950", errors="ignore")}') # Windows 中文編碼
if stderr:
print(f'標準錯誤:\n{stderr.decode("cp950", errors="ignore")}')
communicate() 關鍵特性timeout 參數 (3.3+)，超時拋出 TimeoutExpired，需手動 proc.kill()input 參數可傳送資料給 stdin (配合 stdin=PIPE)
import subprocess
import sys
# 啟動互動式程式 (如 python、bc、自定義協議)
proc = subprocess.Popen(
[sys.executable, '-u'], # -u: 無緩衝模式，關鍵！
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True, # 文字模式
encoding='utf-8'
)
# 發送指令
proc.stdin.write('print("Hello from child")\n')
proc.stdin.flush() # 必須 flush，否則子進程收不到
# 讀取回應 (阻塞直到換行)
response = proc.stdout.readline()
print(f'子進程回應: {response.strip()}')
# 繼續互動...
proc.stdin.write('2 + 3\n')
proc.stdin.flush()
print(f'計算結果: {proc.stdout.readline().strip()}')
# 結束
proc.stdin.close()
proc.wait()
-u / bufsize=0)預設標準輸出為**行緩衝** (終端) 或**塊緩衝** (管道)。子進程輸出未滿緩衝區不會刷新，導致父進程 readline() 永久阻塞。解決：
-u 參數stdbuf -o0、unbuffer)pexpect/pty 模擬終端 (強制行緩衝)
import subprocess
# 實現: dir /w | wc -l (Windows 範例)
child1 = subprocess.Popen(['dir', '/w'], stdout=subprocess.PIPE, shell=True)
child2 = subprocess.Popen(['wc', '-l'], stdin=child1.stdout, stdout=subprocess.PIPE, shell=True)
# 關鍵：父進程關閉對 child1.stdout 的引用，讓 child1 收到 SIGPIPE 正常退出
child1.stdout.close()
output = child2.communicate()[0]
print(f'結果: {output.decode().strip()}')
stdout=PIPE，後一進程 stdin=prev.stdoutprev.stdout.close())，否則前一進程寫入阻塞時無法收到 EOFcommunicate() 獲取最終輸出
import subprocess
import select
import sys
proc = subprocess.Popen(
['ping', 'google.com'], # 持續輸出
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True
)
# 非阻塞輪詢 (Linux/Unix 使用 select)
# Windows 需用 threading + 佇列或 asyncio
while True:
# 檢查是否有資料可讀 (timeout=0.1s)
ready, _, _ = select.select([proc.stdout, proc.stderr], [], [], 0.1)
if proc.stdout in ready:
line = proc.stdout.readline()
if line:
print(f'[OUT] {line.strip()}')
else:
break # EOF
if proc.stderr in ready:
line = proc.stderr.readline()
if line:
print(f'[ERR] {line.strip()}')
# 檢查進程是否結束
if proc.poll() is not None:
# 讀取剩餘輸出
for line in proc.stdout:
print(f'[OUT] {line.strip()}')
break
print(f'進程結束，返回碼: {proc.returncode}')
threading + queue
import subprocess
import threading
import queue
import sys
def reader(stream, q, label):
for line in iter(stream.readline, ''):
q.put((label, line))
stream.close()
proc = subprocess.Popen(
[sys.executable, '-u', 'long_running_script.py'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True
)
q = queue.Queue()
t_out = threading.Thread(target=reader, args=(proc.stdout, q, 'OUT'))
t_err = threading.Thread(target=reader, args=(proc.stderr, q, 'ERR'))
t_out.daemon = True
t_err.daemon = True
t_out.start()
t_err.start()
while proc.poll() is None:
try:
label, line = q.get(timeout=0.1)
print(f'[{label}] {line.strip()}')
except queue.Empty:
pass
# 讀取剩餘
while not q.empty():
label, line = q.get()
print(f'[{label}] {line.strip()}')
import subprocess
# 將輸出直接寫入檔案 (避免記憶體佔用)
with open('app.log', 'w') as log_out, open('app.err', 'w') as log_err:
proc = subprocess.Popen(
['long_task.exe'],
stdout=log_out,
stderr=log_err
)
proc.wait()
# 同時輸出到檔案和管道 (tee 效果) - 需 threading
# 或使用 shell: proc = Popen('cmd 2>&1 | tee log.txt', shell=True)
import subprocess
import signal
import time
proc = subprocess.Popen(['sleep', '100'])
try:
# 等待最多 5 秒
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
print('超時，嘗試優雅終止...')
proc.terminate() # SIGTERM (Windows: CTRL_BREAK_EVENT)
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
print('強制終止...')
proc.kill() # SIGKILL (Windows: TerminateProcess)
proc.wait()
print('已終止')
| 陷阱 | 現象 | 原因 | 解決 |
|---|---|---|---|
| 雙管道死鎖 | 程式卡住不動 | stdout/stderr 管道填滿，子進程寫入阻塞，父進程未讀取 | **必須用 communicate()** 或 threading 並發讀取 |
| 緩衝導致無輸出 | readline() 永久阻塞 | 管道塊緩衝，子進程未 flush | 子進程 -u / stdbuf -o0 / pty / threading 讀取 |
| communicate() 後再次呼叫 | ValueError: I/O operation on closed file | communicate 會關閉管道 | 一次性用 communicate()，或手動管理流 |
| shell=True 安全風險 | 命令注入 | 用戶輸入未轉義進入 shell | 優先用列表 args，避免 shell=True |
| 編碼錯誤 | UnicodeDecodeError / 亂碼 | Windows cp950/gbk vs UTF-8 | 指定 encoding、errors='ignore'/'replace' |
| 殭屍進程 | 資源洩漏 | 父進程未 wait()/communicate() | 總是呼叫 wait() 或 communicate() |
| 大量輸出記憶體爆炸 | MemoryError | communicate() 累積所有輸出 | 流式處理、寫檔案、迴圈 readline |
subprocess.run (Python 3.5+)90% 場景足夠：
import subprocess
# 簡單執行，獲取輸出
result = subprocess.run(
['cmd', 'arg'],
capture_output=True, # 等同 stdout=PIPE, stderr=PIPE
text=True, # 等同 universal_newlines=True
encoding='utf-8',
timeout=30,
check=True # 非零返回碼拋出 CalledProcessError
)
print(result.stdout)
print(result.stderr)
print(result.returncode)
subprocess.Popen 是 Python 進程控制的基石。掌握 communicate() 避免死鎖、stdin=PIPE 雙向互動、管道鏈接關閉中間引用、非阻塞讀取 (threading/select/asyncio)、超時與優雅終止、編碼處理，即可應對從簡單命令執行到複雜進程協作的各種場景。現代代碼優先使用 subprocess.run 簡化常見情況；需要精細控制 (長時間運行、互動式、流式處理、管道鏈接) 時再直接使用 Popen。記住鐵律：**任何雙向管道、雙管道場景，若不使用 communicate()，必須並發讀取 stdout/stderr，否則必死鎖。**
