本文旨在介绍如何使用 Python 并发执行多个子进程并高效捕获它们的输出。通过使用 multiprocessing.pool.ThreadPool，我们可以避免阻塞主线程，从而显著提高程序的整体执行效率，尤其是在需要同时运行大量独立子进程的场景下。
在 Python 中，使用 subprocess 模块可以方便地创建和管理子进程。然而，当需要同时启动并等待多个子进程完成时，传统的串行方式可能会导致效率瓶颈。subprocess.Popen 本身是非阻塞的，这意味着启动子进程后会立即返回，但 proc.communicate() 方法会阻塞，直到子进程执行完毕。如果按照串行方式调用 communicate()，实际上子进程的执行顺序是被强制的，无法充分利用多核 CPU 的并行处理能力。
使用 ThreadPool 并发执行子进程
为了解决这个问题，我们可以使用 multiprocessing.pool.ThreadPool 来并发地执行 communicate() 方法，从而实现真正的并行处理。ThreadPool 允许我们将任务分配给一个线程池，由线程池负责调度和执行这些任务。
以下是一个改进后的代码示例：
立即学习“Python免费学习笔记（深入）”；
import subprocessimport loggingfrom multiprocessing.pool import ThreadPoollog = logging.getLogger(__name__) # 假设已经配置好 loggingdef runShowCommands(cmdTable) -> dict: """ 并发执行 cmdTable 中定义的命令，并返回一个包含命令输出的字典。 """ procOutput = {} # 用于存储命令输出的字典 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) # Add shell=True except Exception as e: log.error(f"Error starting subprocess {cmd}: {e}") procOutput[cmd] = f"Error starting subprocess: {e}" # Store error message to procOutput continue # Skip to the next command # 定义处理子进程输出的函数 def handle_proc_stdout(handle): try: stdout, stderr = procHandles[handle].communicate(timeout=180) procOutput[handle] = stdout.decode("utf-8") # 将 stdout 部分转换为文本 log.debug(f"subprocess returned {handle}") if stderr: log.error(f"subprocess {handle} stderr: {stderr.decode('utf-8')}") except subprocess.TimeoutExpired: log.warning(f"subprocess {handle} timed out") procHandles[handle].kill() procOutput[handle] = "Timeout" except Exception as e: log.error(f"Error communicating with subprocess {handle}: {e}") procOutput[handle] = f"Error communicating: {e}" # Store error message to procOutput # 使用线程池并发执行 communicate threadpool = ThreadPool() threadpool.map(handle_proc_stdout, procHandles.keys()) threadpool.close() threadpool.join() return procOutput
代码解释：
引入 ThreadPool: 首先，我们从 multiprocessing.pool 引入 ThreadPool 类。启动子进程: 循环遍历 cmdTable，使用 subprocess.Popen 启动所有子进程，并将进程句柄存储在 procHandles 字典中。 shell=True 被添加到 subprocess.Popen 调用中。这允许您直接执行包含 shell 命令的字符串，而无需手动拆分命令。定义 handle_proc_stdout 函数: 该函数负责调用 communicate() 方法，获取子进程的输出，并将结果存储在 procOutput 字典中。 函数还包括一个错误处理块，用于捕获超时和任何其他可能发生的异常。如果在与子进程通信时发生错误，它会将错误消息存储在 procOutput 字典中。 此外，它还会记录子进程的标准错误输出 (stderr)。创建 ThreadPool 并执行任务: 创建一个 ThreadPool 实例，并使用 map() 方法将 handle_proc_stdout 函数应用到 procHandles.keys() 上。map() 方法会将 procHandles.keys() 中的每个键作为参数传递给 handle_proc_stdout 函数，并在线程池中并发执行这些函数调用。关闭并等待线程池: 调用 threadpool.close() 方法，防止线程池接受新的任务。然后，调用 threadpool.join() 方法，等待线程池中的所有任务完成。返回结果: 返回包含所有子进程输出的 procOutput 字典。
注意事项：
线程安全： 确保子进程的操作是线程安全的，避免出现竞态条件。资源限制： ThreadPool 的大小应该根据系统的 CPU 核心数和内存资源进行合理配置，避免过度占用资源。错误处理： 在 handle_proc_stdout 函数中添加适当的错误处理机制，例如超时处理和异常捕获，以保证程序的健壮性。日志记录: 添加了日志记录以帮助调试并提供有关子进程执行的洞察力。
总结：
通过使用 ThreadPool，我们可以显著提高并发执行多个子进程的效率。这种方法特别适用于需要同时运行大量独立子进程的场景，例如并行编译、数据处理等。 务必注意线程安全、资源限制和错误处理，以确保程序的稳定性和可靠性。
以上就是并发执行多个 Python 子进程并捕获输出的优化方法的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1371726.html
微信扫一扫 
支付宝扫一扫 
