如何通过Python的subprocess并行执行多个beeline命令并捕获各查询输出

你写的Popen逻辑实际已经并发启动了所有子进程，并不是串行执行，但现有写法存在几个问题：

• wait()和communicate()的顺序不合理：子进程输出内容较多时，stdout/stderr的系统缓冲区会被占满，导致子进程阻塞无法退出，wait()会永久卡住
• 没有将子进程和对应的host绑定，最终拿到输出也无法区分是哪台服务器的执行结果
• beeline命令的-e参数多了一层不必要的引号：用列表形式传命令参数时，subprocess会自动处理转义，额外加的双引号会被识别为SQL语句的一部分，导致语法报错

用concurrent.futures.ThreadPoolExecutor实现，每个线程处理一台服务器的命令执行、输出捕获、异常处理，完全隔离，单台执行失败不会影响其他服务器，还能直接拿到每台对应结果。

import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed

# 配置项
urls = ['host1','host2','host3','host4','host10'] 
userid = "替换为你的用户名"
password = "替换为你的密码"
sql = "select count(*) from DB.TABLENAME;"

def run_beeline(host):
    jdbc_string = f"jdbc:hive2://{host}:1000/;ssl=true"
    # 注意：-e 后面的sql不需要额外加引号
    beeline_cmd = ['beeline','-u',jdbc_string,'-n',userid,'-p',password,'-e', sql]
    try:
        # 执行命令，设置超时防止单台服务器无响应导致整个脚本卡死
        p = subprocess.run(
            beeline_cmd,
            capture_output=True,
            text=True,
            timeout=300 # 超时时间可自行调整，单位为秒
        )
        return {
            "host": host,
            "success": p.returncode == 0,
            "stdout": p.stdout,
            "stderr": p.stderr,
            "return_code": p.returncode
        }
    except Exception as e:
        return {
            "host": host,
            "success": False,
            "error_msg": str(e)
        }

if __name__ == "__main__":
    # 最大并发数设置为服务器数量即可，10台并发无压力
    with ThreadPoolExecutor(max_workers=len(urls)) as executor:
        futures = {executor.submit(run_beeline, host): host for host in urls}
        # 逐个获取执行结果，哪台先跑完就先处理哪台的结果
        for future in as_completed(futures):
            result = future.result()
            host = result["host"]
            if result["success"]:
                print(f"===== {host} 执行成功 =====")
                print("查询输出：", result["stdout"])
            else:
                print(f"===== {host} 执行失败 =====")
                if "error_msg" in result:
                    print("异常信息：", result["error_msg"])
                else:
                    print("错误输出：", result["stderr"])
                    print("返回码：", result["return_code"])

• 完全并行执行，所有服务器的命令同时启动，总耗时等于最慢的那台的执行时间
• 单台执行失败/超时完全不影响其他服务器的执行和结果捕获
• 直接绑定host和对应结果，清晰区分每台的输出
• 内置超时机制，避免某个服务器无响应导致整个脚本卡住
• 规避了缓冲区满导致的死锁问题，subprocess.run会自动处理输出的读取

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