如何通过Python subprocess循环执行Shell命令且不重复创建进程？

方法1：启动交互式Shell持续发送命令 #

可以启动一个交互式cmd.exe进程，通过标准输入持续发送命令，从标准输出读取结果，全程仅创建一次Shell进程，避免重复启动的开销。

示例代码：

import subprocess
import time

# 启动交互式cmd进程，开启行缓冲以便实时获取输出
cmd_process = subprocess.Popen(
    ['cmd.exe'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    bufsize=1
)

def check_process(process_name):
    # 向Shell发送检查命令，必须加换行符触发执行
    cmd_process.stdin.write(f'tasklist | findstr "{process_name}"\n')
    cmd_process.stdin.flush()
    
    # 读取输出，过滤掉命令回显和Shell提示符
    output_lines = []
    while True:
        line = cmd_process.stdout.readline()
        if not line:
            break
        stripped_line = line.strip()
        if stripped_line == f'tasklist | findstr "{process_name}"' or '>' in stripped_line:
            continue
        output_lines.append(stripped_line)
    return '\n'.join(output_lines)

# 循环检查目标进程
while True:
    result = check_process('abc.exe')
    if result:
        print('abc.exe is running.')
        break
    time.sleep(1)  # 间隔1秒检查，避免资源占用过高

# 关闭Shell进程
cmd_process.stdin.write('exit\n')
cmd_process.stdin.flush()
cmd_process.wait()

方法2：直接用Python查询Windows进程（更高效） #

没必要依赖tasklist和findstr这类外部命令，直接通过Python调用Windows API或第三方库查询进程，完全避免Shell和外部进程开销，效率与可靠性更高。

方案2.1：使用第三方库psutil #

先安装库：pip install psutil

import psutil
import time

def is_process_running(process_name):
    for proc in psutil.process_iter(['name']):
        try:
            if proc.info['name'].lower() == process_name.lower():
                return True
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return False

# 循环检查进程状态
while True:
    if is_process_running('abc.exe'):
        print('abc.exe is running.')
        break
    time.sleep(1)

方案2.2：用ctypes调用Windows原生API（无需第三方库） #

import ctypes
import time
from ctypes import wintypes

# 绑定Windows API函数
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
psapi = ctypes.WinDLL('psapi', use_last_error=True)

EnumProcesses = psapi.EnumProcesses
EnumProcesses.argtypes = [wintypes.LPDWORD, wintypes.DWORD, wintypes.LPDWORD]
EnumProcesses.restype = wintypes.BOOL

GetProcessImageFileNameW = psapi.GetProcessImageFileNameW
GetProcessImageFileNameW.argtypes = [wintypes.HANDLE, wintypes.LPWSTR, wintypes.DWORD]
GetProcessImageFileNameW.restype = wintypes.DWORD

OpenProcess = kernel32.OpenProcess
OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
OpenProcess.restype = wintypes.HANDLE

CloseHandle = kernel32.CloseHandle
CloseHandle.argtypes = [wintypes.HANDLE]
CloseHandle.restype = wintypes.BOOL

PROCESS_QUERY_INFORMATION = 0x0400
PROCESS_VM_READ = 0x0010

def is_process_running(process_name):
    process_ids = (wintypes.DWORD * 1024)()
    buffer_size = ctypes.sizeof(process_ids)
    bytes_returned = wintypes.DWORD()
    
    if not EnumProcesses(process_ids, buffer_size, ctypes.byref(bytes_returned)):
        return False
    
    num_processes = bytes_returned.value // ctypes.sizeof(wintypes.DWORD)
    
    for i in range(num_processes):
        pid = process_ids[i]
        h_process = OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid)
        if h_process:
            path_buffer = ctypes.create_unicode_buffer(260)  # MAX_PATH长度
            if GetProcessImageFileNameW(h_process, path_buffer, ctypes.sizeof(path_buffer)):
                # 提取进程文件名（路径最后一段）
                exe_name = path_buffer.value.split('\\')[-1]
                if exe_name.lower() == process_name.lower():
                    CloseHandle(h_process)
                    return True
            CloseHandle(h_process)
    return False

# 循环检查进程
while True:
    if is_process_running('abc.exe'):
        print('abc.exe is running.')
        break
    time.sleep(1)

总结 #

• 若坚持使用Shell命令，方法1可实现单Shell进程重复执行命令，降低开销。
• 更推荐方法2，直接通过Python查询进程，完全规避外部进程开销，效率与稳定性更佳。

内容的提问来源于stack exchange，提问作者user1021