使用线程池并发执行子进程以提高效率-Python教程-PHP中文网

使用线程池并发执行子进程以提高效率

DDD

发布： 2025-09-14 19:15:24

原创

553人浏览过

使用线程池并发执行子进程以提高效率

本文旨在介绍如何使用 Python 的 subprocess 模块并发执行多个子进程，并通过线程池来显著提高程序的执行效率。我们将分析常见的使用 subprocess.Popen 和 .communicate() 方法的场景，并提供使用 ThreadPool 并发等待子进程完成的示例代码。

在使用 subprocess 模块执行多个子进程时，一个常见的误解是 Popen 函数会阻塞程序的执行。实际上，Popen 函数是非阻塞的，它会立即返回一个 Popen 对象，允许程序继续执行。然而，Popen 对象的 communicate() 方法是阻塞的，它会等待子进程执行完毕并返回其输出。如果在循环中依次调用 communicate() 方法，实际上会导致子进程按顺序执行，从而降低程序的效率。

为了解决这个问题，可以使用线程池来并发等待子进程完成。线程池可以创建多个线程，每个线程负责等待一个子进程完成。这样，多个子进程可以同时运行，从而提高程序的执行效率。

下面是一个使用线程池并发等待子进程完成的示例代码：

import subprocess
import logging
from multiprocessing.pool import ThreadPool

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
log = logging.getLogger(__name__)

def runShowCommands(cmdTable) -> dict:
    """return a dictionary of captured output from commands defined in cmdTable."""
    procOutput = {}  # dict to store the output text from show commands
    procHandles = {}
    for cmd, command in cmdTable.items():
        try:
            log.debug(f"running subprocess {cmd} -- {command}")
            procHandles[cmd] = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
        except Exception as e:
            log.error(f"Error launching subprocess {cmd}: {e}")
            continue

    def handle_proc_stdout(handle):
        try:
            proc = procHandles[handle]
            stdout, stderr = proc.communicate(timeout=180)
            procOutput[handle] = stdout.decode("utf-8")  # turn stdout portion into text
            log.debug(f"subprocess returned {handle}")
            if stderr:
                log.error(f"subprocess {handle} returned stderr: {stderr.decode('utf-8')}")
        except subprocess.TimeoutExpired:
            log.error(f"subprocess {handle} timed out")
            proc.kill() # Terminate the process
        except Exception as e:
            log.error(f"Error handling subprocess {handle}: {e}")


    threadpool = ThreadPool()
    threadpool.map(handle_proc_stdout, procHandles.keys())
    threadpool.close()
    threadpool.join()

    return procOutput

if __name__ == '__main__':
    cmdTable = {
        'himom': "echo hi there momma",
        'goodbye': "echo goodbye",
        'date': "date",
        'sleep': "sleep 2 && echo slept"
    }

    output = runShowCommands(cmdTable)

    for cmd, out in output.items():
        print(f"Output from {cmd}:\n{out}")

登录后复制

代码解释：

runShowCommands(cmdTable) 函数：

豆包AI编程
豆包推出的AI编程助手

483

查看详情
- 接受一个字典 cmdTable，其中键是命令的名称，值是要执行的命令字符串。
- 创建一个空字典 procOutput 来存储每个命令的输出。
- 创建一个空字典 procHandles 来存储每个 Popen 对象。
- 循环遍历 cmdTable 中的每个命令：
  - 使用 subprocess.Popen 启动子进程，并将 stdout 和 stderr 重定向到管道。 shell=True 允许直接执行字符串命令，但要注意安全性。
  - 将 Popen 对象存储在 procHandles 字典中，键是命令名称。
- 定义一个内部函数 handle_proc_stdout(handle)：
  - 此函数负责处理单个子进程的输出。
  - 使用 procHandles[handle].communicate(timeout=180) 获取子进程的输出，并设置超时时间为 180 秒。
  - 将输出解码为 UTF-8 字符串，并将其存储在 procOutput 字典中。
  - 记录子进程返回的消息。
  - 处理 TimeoutExpired 异常，如果子进程超时，则记录错误并终止该进程。
  - 处理其他异常，如果发生任何其他错误，则记录错误消息。
- 创建一个 ThreadPool 对象。
- 使用 threadpool.map(handle_proc_stdout, procHandles.keys()) 将 handle_proc_stdout 函数应用于 procHandles 字典中的每个键（命令名称）。这会在线程池中并行执行 handle_proc_stdout 函数。
- 调用 threadpool.close() 以防止向线程池提交更多任务。
- 调用 threadpool.join() 以等待所有线程完成。
- 返回 procOutput 字典。
if __name__ == '__main__': 块：
- 创建一个示例 cmdTable 字典。
- 调用 runShowCommands(cmdTable) 执行命令并获取输出。
- 循环遍历 output 字典，并打印每个命令的输出。

注意事项：

超时处理： communicate(timeout=180) 设置了超时时间，防止子进程无限期运行。如果子进程在指定时间内未完成，将引发 TimeoutExpired 异常，并且该进程将被终止。
错误处理： 代码包含 try...except 块，用于捕获可能发生的异常，例如子进程启动失败或超时。
线程安全： 确保在多线程环境中访问和修改共享资源（例如 procOutput 字典）是线程安全的。在此示例中，由于每个线程都写入不同的键，因此字典的写入操作是线程安全的。
资源限制： 创建过多的线程可能会消耗大量系统资源。线程池的大小应根据系统资源和任务的性质进行调整。
安全性： 使用 shell=True 执行命令时，需要注意命令注入的风险。确保命令字符串来自可信来源，或者对输入进行适当的转义。
日志记录： 使用 logging 模块记录程序的运行状态，方便调试和排错。