Python子进程资源监控：精确测量内存与CPU时间

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

Python子进程资源监控：精确测量内存与CPU时间

2025-12-14

浏览次数：次

返回列表

python子进程资源监控：精确测量内存与cpu时间

本文旨在指导如何在Unix环境下使用Python精确监控子进程的内存占用和CPU时间。我们将探讨使用`subprocess`启动进程，结合`resource`库测量CPU时间，以及`psutil`库跟踪内存使用的最佳实践。重点解决`resource.getrusage`在不当位置调用导致时间统计为零的常见问题，并提供一个结构清晰、功能完整的示例代码。

引言：子进程资源监控的重要性

在数据分析、科学计算或系统管理等领域，经常需要执行外部命令或第三方工具作为子进程。为了评估这些工具的性能、优化资源分配或进行基准测试，精确监控子进程的资源使用情况至关重要。本教程将详细介绍如何利用Python的subprocess、resource和psutil库，在Unix系统上有效地测量子进程的内存占用和CPU时间。

核心库介绍

subprocess: Python标准库，用于创建和管理子进程。它提供了比旧版os.system等函数更强大的功能，能够更好地控制子进程的输入、输出和错误流。
resource: Python标准库，提供了一系列函数来查询和设置系统资源限制。在Unix系统中，它能够获取进程及其子进程的CPU时间、内存使用等详细信息。
psutil: 一个强大的第三方库，用于获取系统和进程的各种信息（CPU、内存、磁盘、网络等）。它提供了一个跨平台的接口，但在本教程中主要用于获取进程的实时内存使用情况。

常见陷阱：resource.getrusage的误用

在使用resource.getrusage(resource.RUSAGE_CHILDREN)来测量子进程的CPU时间时，一个常见的错误是将结束测量点放置在子进程完成之前。resource.RUSAGE_CHILDREN设计用于统计已终止子进程的资源使用情况。这意味着，如果在子进程仍在运行时调用usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)，它将只报告在当前进程生命周期中已经终止的子进程所消耗的资源，而不会包含当前正在运行的子进程。

因此，为了准确获取目标子进程的CPU时间，必须在子进程完全终止并被父进程回收资源后，再调用resource.getrusage(resource.RUSAGE_CHILDREN)。

立即学习“Python免费学习笔记（深入）”；

精确监控子进程资源的使用

以下是一个结合了subprocess、resource和psutil的完整示例，展示了如何正确地监控子进程的内存和CPU时间。

Pippit AI

CapCut推出的AI创意内容生成工具

133 查看详情 Pippit AI

import sys
import os
import subprocess
import resource
import psutil
import time
import datetime

def get_process_memory_info(pid):
    """
    获取指定PID进程的内存使用信息（常驻内存RSS）。
    如果进程不存在，返回None。
    """
    try:
        process = psutil.Process(pid)
        # memory_info().rss 是常驻内存集大小 (Resident Set Size)，以字节为单位
        # 转换为GB
        return process.memory_info().rss / (1024.0 ** 3)
    except psutil.NoSuchProcess:
        return None
    except Exception as e:
        print(f"获取进程 {pid} 内存信息失败: {e}", file=sys.stderr)
        return None

def monitor_subprocess_resources(cmd_list, report_file_path, slice_in_seconds=1):
    """
    监控子进程的CPU时间（用户态和系统态）和内存使用。

    Args:
        cmd_list (list): 包含命令及其参数的列表，例如 ['bioinformatics_tool', 'arg1', 'arg2']。
        report_file_path (str): 报告文件路径，用于保存监控结果。
        slice_in_seconds (int): 内存采样间隔时间（秒）。
    """
    print(f"开始监控命令: {' '.join(cmd_list)}")
    print(f"报告将写入: {report_file_path}")

    # 记录开始时间，用于计算总运行时间（可选，与resource模块无关）
    start_time_wall = time.time()

    # 在子进程启动前获取初始资源使用情况
    # resource.RUSAGE_CHILDREN 统计所有已终止子进程的资源
    # 如果父进程之前没有其他子进程，这里通常是0
    usage_start = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 启动子进程
    # stdout=subprocess.DEVNULL 将标准输出重定向到空设备
    # stderr=subprocess.PIPE 捕获标准错误输出
    try:
        process = subprocess.Popen(
            cmd_list,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            encoding='utf-8'
        )
        pid = process.pid
        print(f"子进程PID: {pid}")
    except FileNotFoundError:
        print(f"错误：命令 '{cmd_list[0]}' 未找到。", file=sys.stderr)
        return
    except Exception as e:
        print(f"启动子进程失败: {e}", file=sys.stderr)
        return

    # 存储内存采样结果
    memory_samples = []

    # 循环检查子进程状态并采样内存
    while process.poll() is None:
        current_memory_gb = get_process_memory_info(pid)
        if current_memory_gb is not None:
            memory_samples.append(current_memory_gb)
        time.sleep(slice_in_seconds)

    # 子进程已终止，检查返回码
    if process.returncode != 0:
        error_output = process.stderr.read()
        print(f"子进程执行失败，返回码: {process.returncode}", file=sys.stderr)
        print(f"错误输出:\n{error_output}", file=sys.stderr)
        sys.exit(f"FAILED: {' '.join(cmd_list)}\n{error_output}")

    # 在子进程终止后获取最终资源使用情况
    # 此时 resource.RUSAGE_CHILDREN 会包含刚刚终止的子进程的资源
    usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)
    end_time_wall = time.time() # 记录结束时间

    # 计算CPU时间
    cpu_time_user = usage_end.ru_utime - usage_start.ru_utime
    cpu_time_system = usage_end.ru_stime - usage_start.ru_stime
    total_cpu_time = cpu_time_user + cpu_time_system
    wall_clock_time = end_time_wall - start_time_wall

    # 写入报告文件
    with open(report_file_path, "w") as outrepfp:
        outrepfp.write(f"Command: {' '.join(cmd_list)}\n")
        outrepfp.write(f"Wall Clock Time: {wall_clock_time:.4f} seconds\n")
        outrepfp.write(f"User CPU Time: {cpu_time_user:.4f} seconds\n")
        outrepfp.write(f"System CPU Time: {cpu_time_system:.4f} seconds\n")
        outrepfp.write(f"Total CPU Time: {total_cpu_time:.4f} seconds\n")
        outrepfp.write(f"Memory Usage (GB) Samples: {memory_samples}\n")
        if memory_samples:
            outrepfp.write(f"Peak Memory (GB): {max(memory_samples):.4f}\n")
        else:
            outrepfp.write("Peak Memory (GB): N/A (no memory samples collected)\n")

    print("\n监控完成，报告已生成。")
    print(f"Wall Clock Time: {wall_clock_time:.4f} seconds")
    print(f"User CPU Time: {cpu_time_user:.4f} seconds")
    print(f"System CPU Time: {cpu_time_system:.4f} seconds")
    print(f"Peak Memory (GB): {max(memory_samples) if memory_samples else 'N/A'}")


# --- 示例用法 ---
if __name__ == "__main__":
    # 模拟一个长时间运行的命令
    # 例如：'sleep 5' 会运行5秒
    # 对于生物信息学工具，替换为你的实际命令
    # 例如: bioinformatics_tool = "bwa"
    #      setups = "mem -t 4"
    #      resultdir = "output.sam"
    #      inputs = "input.fastq"
    #      cmd = [bioinformatics_tool, setups, "--tblout", resultdir, inputs]

    # 示例1: 简单的sleep命令
    print("--- 运行示例 1: sleep 5 ---")
    mock_command_sleep = ['sleep', '5'] # 模拟一个运行5秒的命令
    report_file_sleep = "report_sleep.txt"
    monitor_subprocess_resources(mock_command_sleep, report_file_sleep, slice_in_seconds=1)
    print("-" * 30)

    # 示例2: 模拟一个简单的Python脚本作为子进程
    # 创建一个模拟的Python脚本文件
    with open("mock_script.py", "w") as f:
        f.write("""
import time
import sys
import os
print("Mock script started.")
# 模拟内存使用增长 (并非实际内存分配，仅为示例)
data = []
for i in range(10):
    time.sleep(0.5)
    # 实际内存使用可以通过分配大对象来模拟，这里仅为示意
    # data.append(os.urandom(1024 * 1024)) # 每次增加1MB，但会很快耗尽内存
    print(f"Mock script working... {i+1}s")
    sys.stdout.flush() # 确保输出及时显示
print("Mock script finished.")
""")
    print("--- 运行示例 2: mock_script.py ---")
    mock_command_python_script = [sys.executable, 'mock_script.py']
    report_file_python_script = "report_python_script.txt&quot;
    monitor_subprocess_resources(mock_command_python_script, report_file_python_script, slice_in_seconds=0.5)
    os.remove("mock_script.py") # 清理模拟脚本
    print("-" * 30)

代码解析与注意事项

get_process_memory_info(pid) 函数:
- 此函数使用psutil.Process(pid)获取特定进程对象，然后通过process.memory_info().rss获取其常驻内存集大小（Resident Set Size, RSS）。RSS是进程实际占用物理内存的部分。
- 返回值为GB，便于阅读。
- 增加了错误处理，以防进程在采样期间终止。
- 与原始问题的区别: 原始问题中的get_memory_info函数获取的是系统总内存信息，而非特定子进程的内存。本教程已将其修改为获取指定pid子进程的精确内存使用，这更符合“监控子进程”的目标。
monitor_subprocess_resources 函数:
- subprocess.Popen: 使用subprocess.Popen启动子进程，并指定stdout=subprocess.DEVNULL将标准输出丢弃，stderr=subprocess.PIPE捕获标准错误，以便在进程失败时进行检查。encoding='utf-8'确保错误信息能正确解码。
- 内存采样循环: while process.poll() is None:循环会持续检查子进程是否仍在运行。只要子进程未终止，就会调用get_process_memory_info进行内存采样，并暂停slice_in_seconds秒。
- resource.getrusage的正确位置:
  - usage_start = resource.getrusage(resource.RUSAGE_CHILDREN)在子进程启动前调用，用于建立一个基线。
  - usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)在while循环结束后，即子进程完全终止后调用。这是确保ru_utime和ru_stime包含目标子进程CPU时间的关键。
- CPU时间计算:
  - cpu_time_user: 用户态CPU时间，表示进程在用户模式下执行指令所花费的时间。
  - cpu_time_system: 系统态CPU时间，表示进程在内核模式下执行系统调用所花费的时间。
  - 这些时间都是子进程（及其后代）消耗的累计CPU时间。
- 墙钟时间 (Wall Clock Time): 通过time.time()记录开始和结束时间，计算出实际经过的总时间，这与CPU时间不同，因为它包含了等待I/O、调度等非CPU密集型操作的时间。
- 错误处理: 检查process.returncode，如果非零，则表示进程执行失败，并打印错误信息。
- 报告输出: 将所有收集到的数据写入指定的报告文件，包括命令、墙钟时间、用户/系统CPU时间、内存采样列表以及峰值内存。
平台兼容性:
- resource模块是Unix特有的。在Windows系统上，尝试导入resource会引发ImportError。因此，此解决方案仅适用于Unix-like系统（Linux, macOS等）。
- psutil是跨平台的，但其获取内存的方式在不同系统上可能略有差异，不过基本功能是通用的。
采样频率: slice_in_seconds参数决定了内存采样的频率。更小的值会提供更精细的内存使用曲线，但也会增加父进程的CPU开销。需要根据实际需求和子进程的运行特性进行权衡。

总结

通过本教程，我们学习了如何在Python中利用subprocess、resource和psutil库，在Unix环境下对子进程的内存和CPU时间进行精确监控。核心要点在于理解resource.getrusage(resource.RUSAGE_CHILDREN)的工作机制，确保在子进程终止后才进行最终的资源统计，以避免时间统计为零的常见问题。结合psutil的实时内存采样，我们可以获得子进程运行期间全面的性能数据，这对于性能分析和优化具有重要意义。

以上就是Python子进程资源监控：精确测量内存与CPU时间的详细内容，更多请关注其它相关文章！

# python # 寺庙营销推广 # 吉林视频网站优化方案 # 百度营销推广顾问面试 # 集大 # 就会 # 这是 # 是一个 # 都是 # 的是 # 为零 # 错误信息 # 仅为 # linux # windows # app # 字节 # 工具 # mac # ai # unix # macos # win # 第三方 # 诸暨网站优化好不好用 # 网站推广优化开发建设 # 南宁seo网站优化排名 # 铁耙云网站建设 # 郑州营销推广活动有哪些 # 301跳转很多 seo # 湖北seo推广教程