Python中subprocess.Popen()返回None，需维持进程并单独计时目标查询

首先明确：subprocess.Popen()不会返回None，如果你的process变量为None，大概率是代码存在其他问题——比如Popen调用时抛出未捕获异常导致变量未正确赋值，或是误将Popen写成了subprocess.run，或是后续代码将process重新赋值为None。可以通过添加异常捕获和打印验证：

from subprocess import Popen, PIPE

cmd = ["/full/path/to/binary","--arg1"]
try:
    process = Popen(cmd, stdin=PIPE, stdout=None)
    print(f"Process实例: {process}")  # 若正常会输出<Popen: returncode: None ...>
except Exception as e:
    print(f"启动进程出错: {e}")

针对你需要复用同一CLI进程、仅统计目标查询耗时的需求，以下是可靠实现方案（以类似MySQL的交互式CLI为例）：

正确实现代码 #

from subprocess import Popen, PIPE
import time
from tqdm import trange

# 配置参数
dbname = "your_db"
forced_plan = "your_plan"
query_text = "SELECT * FROM your_table;"
runs = 10
cli_path = "/full/path/to/mysql"
cli_args = ["--user=your_user", "--password=your_pass"]  # 替换为你的CLI启动参数

# 1. 启动CLI进程，绑定标准输入输出（文本模式避免手动编解码）
process = Popen(
    [cli_path] + cli_args,
    stdin=PIPE,
    stdout=PIPE,
    stderr=PIPE,
    text=True
)

# 2. 执行初始化命令（复用会话状态）
init_cmds = [
    f"use {dbname};",
    "set optimizer_switch='hypergraph_optimizer=on';",
    f"SET forced_plan='{forced_plan}';"
]
for cmd in init_cmds:
    process.stdin.write(cmd + "\n")
process.stdin.flush()

# 读取初始化输出直到CLI提示符（比如mysql>），确认初始化完成
while True:
    line = process.stdout.readline()
    if "mysql>" in line:  # 替换为你的CLI实际提示符
        break

# 3. 循环执行查询并仅统计目标查询耗时
runtimes = []
for _ in trange(runs):
    # 发送查询前开始计时
    start = time.time()
    
    # 发送查询命令
    process.stdin.write(query_text + "\n")
    process.stdin.flush()
    
    # 读取查询输出直到提示符，确认查询完成
    output_lines = []
    while True:
        line = process.stdout.readline()
        if not line:
            break  # 进程意外退出
        output_lines.append(line)
        if "mysql>" in line:
            break
    
    # 停止计时并记录
    end = time.time()
    runtimes.append(end - start)
    
    # 可选：打印查询结果
    print("".join(output_lines))

# 4. 关闭进程
process.stdin.write("exit;\n")
process.stdin.flush()
process.wait()

# 输出统计结果
print(f"总运行次数: {runs}")
print(f"平均耗时: {sum(runtimes)/len(runtimes):.4f} 秒")

关键注意事项 #

• 必须读取stdout：如果CLI进程的输出缓冲区满，会导致进程阻塞，无法继续接收输入。通过读取到提示符的方式，既可以确认命令执行完成，也能避免缓冲区阻塞。
• 使用text=True：Python3.7+支持的参数，直接处理字符串而非字节流，简化交互逻辑。
• 会话状态复用：所有初始化命令在同一个进程中执行，后续查询可以继承这些配置（如数据库选择、优化器设置）。

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