当前位置：首页 > 文章列表 > 文章 > python教程 > Python监控进程资源使用方法

Python监控进程资源使用方法

2025-12-29 17:13:04 0浏览收藏

大家好，今天本人给大家带来文章《Python监控子进程内存与CPU使用教程》，文中内容主要涉及到，如果你对文章方面的知识点感兴趣，那就请各位朋友继续看下去吧~希望能真正帮到你们，谢谢！

Python在Unix环境下监控子进程的内存与CPU时间使用教程

本教程详细阐述了如何在Unix系统下，利用Python的`subprocess`模块执行外部程序，并结合`resource`和`psutil`库精确监控其CPU时间（用户态与系统态）及内存使用情况。文章重点讲解了`resource.getrusage`的正确使用时机，以避免时间测量不准确的问题，并提供了基于进程ID的内存实时采样方法，确保对子进程性能进行全面而准确的评估。

在生物信息学工具开发、系统性能测试或任何需要执行外部命令并评估其效率的场景中，精确监控子进程的资源使用至关重要。本教程将指导您如何使用Python在Unix环境下实现这一目标，重点关注CPU时间（用户态和系统态）以及内存使用。

核心工具简介

我们将主要使用以下Python库：

subprocess: 用于启动新的应用程序或命令。
resource: 提供系统资源管理功能，包括获取进程的CPU时间。
psutil: 一个跨平台的库，用于检索正在运行的进程和系统利用率（CPU、内存、磁盘、网络等）的信息。
time: 用于在内存采样之间引入延迟。

监控CPU时间

resource模块提供了一个getrusage函数，可以用来获取进程及其子进程的资源使用情况。要准确测量由subprocess.Popen启动的子进程的CPU时间，关键在于getrusage的调用时机。

resource.getrusage的正确用法

resource.getrusage(resource.RUSAGE_CHILDREN)会返回所有已终止子进程所使用的资源总和。这意味着，如果您在子进程完成之前调用它来获取结束时的资源使用情况，它将不会包含当前仍在运行的子进程的数据。

常见错误： 在子进程启动后立即调用usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)，而子进程还在后台运行，会导致ru_utime和ru_stime显示为0，因为它只统计了在调用usage_end之前已经终止的子进程。

正确做法：

在启动子进程之前记录初始资源使用情况。
等待子进程完全终止。
在子进程终止后记录最终资源使用情况。

这样，最终的usage_end将包含由subprocess.Popen启动的子进程所消耗的全部CPU时间。

CPU时间类型：

ru_utime: 用户态CPU时间（user CPU time），表示进程在用户模式下执行指令所花费的时间。
ru_stime: 系统态CPU时间（system CPU time），表示进程在内核模式下执行系统调用所花费的时间。

示例代码（CPU时间部分）

import subprocess
import resource
import sys
import time

def measure_cpu_time(command_list):
    """
    测量子进程的CPU时间（用户态和系统态）。
    """
    print(f"Executing command: {' '.join(command_list)}")

    # 1. 启动前获取资源使用情况（包括已终止的子进程，通常为0）
    usage_start = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 2. 启动子进程
    # subprocess.DEVNULL 避免子进程输出到控制台
    # subprocess.PIPE 捕获标准错误，以便在失败时打印
    process = subprocess.Popen(command_list, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, encoding='utf-8')

    # 3. 等待子进程完成
    # process.wait() 会阻塞直到子进程终止
    # 或者使用 process.poll() 循环等待，以便在循环中进行其他监控（如内存）
    process.wait()

    # 4. 子进程结束后获取资源使用情况
    usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 5. 检查子进程退出码
    if process.returncode != 0:
        error_message = process.stderr.strip()
        sys.exit(f"FAILED: {' '.join(command_list)}\n{error_message}")

    # 6. 计算CPU时间
    cpu_time_user = usage_end.ru_utime - usage_start.ru_utime
    cpu_time_system = usage_end.ru_stime - usage_start.ru_stime

    print(f"User CPU Time: {cpu_time_user:.6f} seconds")
    print(f"System CPU Time: {cpu_time_system:.6f} seconds")
    return cpu_time_user, cpu_time_system

if __name__ == "__main__":
    # 示例：运行一个模拟耗时任务的命令
    # 例如：`sleep 5` 会等待5秒，但CPU时间很低
    # 更能体现CPU时间的命令可以是计算密集型任务，如 `dd if=/dev/zero of=/dev/null count=1000000 bs=1K`
    # 或者一个简单的Python脚本
    # command_to_run = ["sleep", "5"]
    command_to_run = ["python", "-c", "import time; sum(range(10**7)); time.sleep(1); sum(range(10**7))"]
    measure_cpu_time(command_to_run)

监控内存使用

psutil库提供了强大的功能来获取系统和进程的详细信息。要监控特定子进程的内存使用，我们需要获取其进程ID（PID），然后使用psutil.Process(pid)来查询该进程的内存信息。

psutil获取进程内存

psutil.Process(pid).memory_info()会返回一个包含多种内存指标的命名元组，其中：

rss (Resident Set Size): 进程实际使用的物理内存大小，是衡量进程内存消耗最常用的指标。
vms (Virtual Memory Size): 进程占用的虚拟内存大小。

为了实时监控，我们需要在子进程运行期间定期采样。

示例代码（内存部分）

import psutil
import time

def get_process_memory_usage(pid):
    """
    获取指定进程的RSS内存使用量（以GB为单位）。
    """
    try:
        process = psutil.Process(pid)
        return process.memory_info().rss / (1024.0 ** 3) # 转换为GB
    except psutil.NoSuchProcess:
        # 进程可能已经结束
        return 0.0
    except psutil.AccessDenied:
        # 权限不足，无法获取进程信息
        print(f"Warning: Access denied for process {pid}. Cannot get memory info.", file=sys.stderr)
        return 0.0

def monitor_memory_snapshots(process_obj, slice_in_seconds=1):
    """
    实时监控并收集子进程的内存快照。
    """
    memory_snapshots = []
    # 循环直到子进程结束
    while process_obj.poll() is None:
        mem_gb = get_process_memory_usage(process_obj.pid)
        if mem_gb > 0: # 只有当成功获取到内存信息时才记录
            memory_snapshots.append(mem_gb)
        time.sleep(slice_in_seconds)
    return memory_snapshots

# 注意：此函数不能单独运行，需要与subprocess结合
# 完整的集成示例将在下一节提供

重要提示： 原始问题中的get_memory_info函数调用的是psutil.virtual_memory()，这获取的是系统整体的内存使用情况，而非单个子进程。如果目标是监控子进程，务必使用psutil.Process(pid)来获取特定进程的内存信息。

整合CPU时间与内存监控

现在，我们将上述方法整合到一个完整的函数中，用于全面监控子进程的性能。

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

def get_process_memory_usage(pid):
    """
    获取指定进程的RSS内存使用量（以GB为单位）。
    """
    try:
        process = psutil.Process(pid)
        return process.memory_info().rss / (1024.0 ** 3) # 转换为GB
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        return 0.0

def monitor_subprocess_performance(command_list, report_file_path="performance_report.txt", slice_in_seconds=1):
    """
    监控子进程的CPU时间（用户态和系统态）及内存使用情况。
    """
    print(f"Executing command: {' '.join(command_list)}")

    # 1. 启动前获取资源使用情况
    usage_start = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 2. 启动子进程
    process = subprocess.Popen(command_list, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, encoding='utf-8')

    # 3. 实时监控内存使用
    memory_snapshots = []
    # 循环直到子进程结束
    while process.poll() is None:
        mem_gb = get_process_memory_usage(process.pid)
        if mem_gb > 0:
            memory_snapshots.append(mem_gb)
        time.sleep(slice_in_seconds)

    # 4. 子进程结束后获取资源使用情况
    usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)

    # 5. 检查子进程退出码
    if process.returncode != 0:
        error_message = process.stderr.strip()
        print(f"ERROR: Subprocess failed with code {process.returncode}:\n{error_message}", file=sys.stderr)
        sys.exit(f"FAILED: {' '.join(command_list)}\n{error_message}")

    # 6. 计算CPU时间
    cpu_time_user = usage_end.ru_utime - usage_start.ru_utime
    cpu_time_system = usage_end.ru_stime - usage_start.ru_stime

    # 7. 写入报告文件
    with open(report_file_path, "w") as f:
        f.write(f"Command: {' '.join(command_list)}\n")
        f.write(f"User CPU Time: {cpu_time_user:.6f} seconds\n")
        f.write(f"System CPU Time: {cpu_time_system:.6f} seconds\n")
        f.write(f"Peak RSS Memory: {max(memory_snapshots) if memory_snapshots else 0:.3f} GB\n")
        f.write(f"Memory Snapshots (RSS in GB): {memory_snapshots}\n")

    print(f"Monitoring complete. Report saved to {report_file_path}")
    print(f"User CPU Time: {cpu_time_user:.6f} seconds")
    print(f"System CPU Time: {cpu_time_system:.6f} seconds")
    print(f"Peak RSS Memory: {max(memory_snapshots) if memory_snapshots else 0:.3f} GB")
    print(f"Memory Snapshots (RSS in GB): {memory_snapshots}")

if __name__ == "__main__":
    # 示例：运行一个模拟耗时且有一定CPU和内存使用的命令
    # 这个Python脚本会进行一些计算并暂停，模拟真实工具的行为
    # 注意：在某些系统上，Python解释器本身也会消耗一定的CPU和内存
    command_to_run = [
        sys.executable, "-c",
        "import time, os; "
        "data = [i*i for i in range(10**6)]; "  # 模拟内存分配和一些计算
        "time.sleep(2); " # 暂停2秒
        "data2 = [x+1 for x in data]; " # 更多计算
        "time.sleep(1); " # 暂停1秒
        "print('Subprocess finished calculation and sleeping.');"
    ]

    # 您也可以替换为您的生物信息学工具命令，例如：
    # command_to_run = ["your_bioinformatics_tool", "--input", "file.fa", "--output", "result.txt"]

    monitor_subprocess_performance(command_to_run, "tool_performance_report.txt", slice_in_seconds=1)

    print("\n--- Testing a command with minimal CPU/memory (e.g., sleep) ---")
    monitor_subprocess_performance(["sleep", "3"], "sleep_performance_report.txt", slice_in_seconds=1)

注意事项与最佳实践

resource.RUSAGE_CHILDREN的理解: 它统计的是所有已终止子进程的资源。因此，usage_end必须在目标子进程终止后获取，才能包含该子进程的资源使用情况。
内存采样频率: SLICE_IN_SECONDS决定了内存采样的频率。更小的值会提供更精细的内存使用曲线，但也会增加监控本身的开销。根据实际需求进行调整。
内存指标选择: psutil.Process(pid).memory_info().rss通常是衡量进程实际物理内存使用最相关的指标。vms（虚拟内存）可能包含未实际使用的内存。
错误处理: 务必检查子进程的returncode，并在非零时捕获并打印stderr中的错误信息，这对于调试外部工具至关重要。
subprocess.Popen参数:
- stdout=subprocess.DEVNULL: 将子进程的标准输出重定向到空设备，避免污染控制台。
- stderr=subprocess.PIPE: 捕

到这里，我们也就讲完了《Python监控进程资源使用方法》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！