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

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时间。

Text Mark

处理文本内容的AI助手

113 查看详情 Text Mark

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"
    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 # windows # linux # 的是 # 周到的福州Seo排名 # 兖州营销推广效果好 # 唐山网站建设及优化 # 集大 # 就会 # 这是 # 是一个 # 都是 # 为零 # 第三方 # app # 字节 # 工具 # mac # ai # unix # macos # win # 仅为 # 错误信息 # seo内容优化原创 # 7月地产营销推广活动 # 汝南推广网站多少钱一个 # 博罗广告网站建设 # 账号包含营销推广信息吗 # 嘉善首页推广营销 # 阿里云seo待遇

相关栏目：【 Google疑问12 】【 Facebook疑问10 】【优化推广96088 】【技术知识133117 】【 IDC资讯59369 】【网络运营7196 】【 IT资讯61894 】

2025-12-14

SQL多表关联如何理解_核心原理解析助你掌握关键方法【教程】 SQL多表连接如何理解_JOIN关联关系详细步骤解析【指导】 SQL多表连接结果异常怎么办_JOIN条件排查方法解析【指导】 SQL索引下推是什么_ICP机制性能提升原理【教程】 SQL跨表统计怎么写_重要技巧总结提升查询效率【技巧】 SQL注入如何防护_完整逻辑拆解助力系统化掌握【技巧】 SQL上传文件信息建模方法_SQL存储文件元数据方案 SQL多语种存储方案设计_SQL字符集选择策略 SQL字符串处理如何编写_重要技巧总结提升查询效率【教学】 SQL去重查询怎么实现_优化思路讲解帮助高效处理数据【技巧】 SQL统计复购用户怎么做_多次行为分析SQL思路【教程】 SQL字符串处理如何编写_优化思路讲解帮助高效处理数据【教程】 SQL动态字段解析技巧_SQL解析JSON多层字段 SQL线上慢SQL如何治理_规范与工具实践总结【技巧】 SQL表结构如何演进_兼容老数据设计思路【教程】 SQL排序规则如何设置_ORDERBY排序原理与性能说明【指导】 SQL去重查询怎么实现_真实案例解析强化复杂查询思维【教学】 SQL按字段范围分批处理_SQL避免长时间锁表 SQL统计不同字段组合数量_多列聚合查询技巧【技巧】 SQL多字段去重怎么处理_GROUPBY去重思路说明【教学】

了解您产品搜索量及市场趋势，制定营销计划

同行竞争及网站分析保障您的广告效果

点击免费数据支持

提交您的需求，1小时内享受我们的专业解答。

运城市盐湖区信雨科技有限公司

运城市盐湖区信雨科技有限公司是一家深耕海外推广领域十年的专业服务商，作为谷歌推广与Facebook广告全球合作伙伴，聚焦外贸企业出海痛点，以数字化营销为核心，提供一站式海外营销解决方案。公司凭借十年行业沉淀与平台官方资源加持，打破传统外贸获客壁垒，助力企业高效开拓全球市场，成为中小企业出海的可靠合作伙伴。