Python 学习笔记(第十一期)——运维自动化(上·后篇):进程级监控与子进程管理——psutil进阶

知识衔接:

Python 学习笔记(第十期)——运维自动化(上·前篇):系统基础监控——CPU内存磁盘网络

更多Python知识:Python基础入门

我的主页:AOwhisky,这里有更多运维系统性知识整理和其他有趣内容,欢迎与我一起探讨学习~

📖 引言:从“系统整体”到“每一个员工”

上一期我们学习了如何用 psutil 给服务器做“全身体检”——查看 CPU 使用率、内存剩余量、磁盘空间、网络流量。这些都是系统整体层面的监控指标,相当于了解一家公司的整体财务状况:总营收、总支出、利润率。

但运维工作远不止于此。当服务器 CPU 突然飙升到 100%,你只知道“整体出问题了”,却不知道是哪个程序导致的。这就需要从“公司整体财报”下沉到“每一个员工的工作状态”——哪个进程(员工)在疯狂消耗 CPU?哪个进程占用了大量内存?哪个进程在不停地读写磁盘?

进程监控就是解决这个问题的关键能力。本期我们将深入学习 psutil 的进程管理功能,掌握如何列出所有进程、获取进程的详细资源使用情况、监控进程状态,并结合 subprocess.Popen 实现系统命令的调用与输出获取。学完本期,你将能够精准定位“罪魁祸首”进程,让运维工作从“被动救火”升级为“主动防控”。

— Compiled and Authored by Whisky — July 19 th, 2026

📑 本期目录

  1. 进程管理概述(什么是进程、为什么需要进程监控)
  2. 获取进程列表(pids()process_iter()
  3. 进程详细信息采集(Process 类详解)
  4. 进程资源监控实战(CPU/内存/IO 使用率统计)
  5. 进程过滤与排序(找出 Top N 资源消耗进程)
  6. 子进程管理(subprocess.Popen 用法详解)
  7. 综合实战:进程资源 Top N 监控脚本
  8. 总结与知识点一览表

正文

一、进程管理概述

1.1 什么是进程

进程(Process)是程序在操作系统中的一次执行实例。每当你启动一个程序(如 Nginx、MySQL、Python 脚本),操作系统就会为该程序创建一个或多个进程。

💡 白话理解:程序是写在硬盘上的“剧本”(静态的代码文件),进程是演员在舞台上“表演”这个剧本(动态的运行实体)。同一个剧本可以被多个演员同时表演——就像你可以同时运行多个 Python 脚本,每个脚本都是一个独立的进程。

1.2 为什么需要进程监控
场景 问题 进程监控的作用
CPU 飙升 整体 CPU 使用率 100%,但不知道是哪个程序导致的 找出 CPU 使用率最高的进程,定位问题程序
内存泄漏 内存使用量持续增长,最终导致 OOM(Out of Memory) 监控各进程的内存增长趋势,提前发现泄漏
僵尸进程 系统出现大量 defunct 进程,消耗进程表资源 检测进程状态,及时发现僵尸进程
服务存活 关键服务(如数据库)是否还在运行 定期检查进程是否存在,实现自动告警或重启

二、获取进程列表

2.1 使用 pids() 获取所有进程 PID

psutil.pids() 返回当前系统中所有进程的 PID(进程 ID)列表。

import psutil

# 获取所有进程 PID
all_pids = psutil.pids()
print(f"当前系统共有 {len(all_pids)} 个进程")
print(f"前 10 个 PID:{all_pids[:10]}")

运行结果

当前系统共有 145 个进程
前 10 个 PID:[1, 2, 3, 4, 5, 6, 7, 9, 11, 12]
2.2 使用 process_iter() 迭代进程对象

pids() 返回的是 PID 列表,如果需要获取每个进程的详细信息,通常使用 process_iter() 更高效——它直接返回进程对象的迭代器,避免逐个调用 psutil.Process(pid) 造成的性能开销。

import psutil

# 方式一:使用 process_iter() 一次性获取所有进程对象
print("使用 process_iter() 遍历所有进程:")
for proc in psutil.process_iter(['pid', 'name', 'status']):
    try:
        print(f"  PID: {proc.info['pid']}, 名称: {proc.info['name']}, 状态: {proc.info['status']}")
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        # 进程可能在遍历过程中结束,或者无权限访问
        pass

运行结果

使用 process_iter() 遍历所有进程:
  PID: 1, 名称: systemd, 状态: running
  PID: 2, 名称: kthreadd, 状态: running
  PID: 3, 名称: rcu_gp, 状态: running
  PID: 4, 名称: rcu_par_gp, 状态: running
  ...

process_iter()pids() 的对比

对比维度 psutil.pids() psutil.process_iter()
返回类型 PID 列表(整数) 进程对象迭代器
使用方式 需逐个实例化 Process(pid) 直接获取进程对象
性能 较慢(多次系统调用) 较快(一次系统调用获取多字段)
推荐场景 仅需 PID 列表 需要进程详细信息的场景

三、进程详细信息采集——Process 类详解

psutil.Process(pid) 返回一个进程对象,包含了该进程的全部信息。下面逐一介绍最常用的方法和属性。

3.1 创建进程对象
import psutil

# 以 PID 1(systemd/init 进程)为例
p = psutil.Process(1)

print(f"进程对象:{p}")
3.2 进程基本信息
import psutil

p = psutil.Process(1)

print(f"进程名称:{p.name()}")           # 进程的可执行文件名
print(f"进程 PID:{p.pid}")              # 进程 ID
print(f"父进程 PID:{p.ppid()}")         # 父进程 ID
print(f"进程状态:{p.status()}")          # running/sleeping/zombie 等
print(f"进程创建时间:{p.create_time()}") # 时间戳
print(f"进程可执行路径:{p.exe()}")       # 二进制文件路径
print(f"进程工作目录:{p.cwd()}")          # 当前工作目录
print(f"进程命令行参数:{p.cmdline()}")   # 启动时的完整命令

运行结果

进程名称:systemd
进程 PID:1
父进程 PID:0
进程状态:running
进程创建时间:1739108289.75
进程可执行路径:/usr/lib/systemd/systemd
进程工作目录:/
进程命令行参数:['/usr/lib/systemd/systemd', '--switched-root', '--system', '--deserialize=21']

进程状态常见值

状态值 含义 运维含义
running 正在运行或等待运行 正常状态
sleeping 可中断睡眠(等待事件) 正常状态,大多数进程处于此状态
disk-sleep 不可中断睡眠(等待 I/O) 可能 I/O 瓶颈
zombie 僵尸进程(已终止但未回收) 需检查父进程是否正确回收
stopped 被暂停(如被 Ctrl+Z 挂起) 进程被调试或挂起
dead 已终止 应被回收

⚠️ 关键指标zombie 状态的进程数量持续增长,说明父进程没有正确调用 wait() 回收子进程,可能存在问题。

3.3 进程 CPU 信息
import psutil

p = psutil.Process(1)

# CPU 时间(累计值)
cpu_times = p.cpu_times()
print(f"CPU 时间 - 用户态:{cpu_times.user:.2f}s")
print(f"CPU 时间 - 系统态:{cpu_times.system:.2f}s")

# 进程 CPU 使用率(需要采样间隔)
# 注意:第一次调用返回 0.0,需要间隔后再调用才能得到有意义的值
print(f"进程 CPU 使用率(瞬时):{p.cpu_percent(interval=0.1)}%")

# CPU 亲和度(进程被绑定到哪些 CPU 核心)
try:
    print(f"CPU 亲和度:{p.cpu_affinity()}")
except AttributeError:
    # Windows 不支持 cpu_affinity
    print("CPU 亲和度:当前系统不支持")

运行结果

CPU 时间 - 用户态:0.03s
CPU 时间 - 系统态:0.04s
进程 CPU 使用率(瞬时):0.0%
CPU 亲和度:[0, 1]

💡 实战技巧cpu_percent(interval=0.1) 会阻塞 0.1 秒进行采样,返回该进程在这段时间内的 CPU 使用率。如果对实时性要求不高,建议使用 interval=1 获取更稳定的值。

3.4 进程内存信息
import psutil

p = psutil.Process(1)

# 内存详细信息
mem_info = p.memory_info()
print(f"RSS(常驻内存):{mem_info.rss / (1024**2):.2f} MB")
print(f"VMS(虚拟内存):{mem_info.vms / (1024**2):.2f} MB")
print(f"共享内存:{mem_info.shared / (1024**2):.2f} MB")
print(f"文本段大小:{mem_info.text / (1024**2):.2f} MB")
print(f"库大小:{mem_info.lib / (1024**2):.2f} MB")
print(f"数据段大小:{mem_info.data / (1024**2):.2f} MB")
print(f"脏页大小:{mem_info.dirty / (1024**2):.2f} MB")

# 进程内存占比
print(f"内存使用率:{p.memory_percent():.2f}%")

运行结果

RSS(常驻内存):12.75 MB
VMS(虚拟内存):21.08 MB
共享内存:10.62 MB
文本段大小:0.34 MB
库大小:0.00 MB
数据段大小:1.88 MB
脏页大小:0.00 MB
内存使用率:0.33%

内存字段说明

字段 全称 含义 运维意义
RSS Resident Set Size 物理内存中实际占用的内存页 最重要的指标,反映进程实际占用的物理内存
VMS Virtual Memory Size 虚拟内存总量(含未分配的地址空间) 反映进程申请的总虚拟地址空间
shared Shared Memory 与其他进程共享的内存 共享库占用的内存
text Text Segment 可执行代码段大小 进程代码的大小
data Data Segment 数据段大小 进程数据(堆/栈)的大小
dirty Dirty Pages 已修改但未写回磁盘的页 内存回收时需关注的指标
3.5 进程 I/O 信息
import psutil

p = psutil.Process(1)

try:
    io_counters = p.io_counters()
    print(f"读次数:{io_counters.read_count}")
    print(f"写次数:{io_counters.write_count}")
    print(f"读字节数:{io_counters.read_bytes / (1024**2):.2f} MB")
    print(f"写字节数:{io_counters.write_bytes / (1024**2):.2f} MB")
    print(f"读字符数:{io_counters.read_chars / (1024**2):.2f} MB")
    print(f"写字符数:{io_counters.write_chars / (1024**2):.2f} MB")
except psutil.AccessDenied:
    print("无权限访问该进程的 I/O 信息")

运行结果

读次数:950
写次数:28
读字节数:0.00 MB
写字节数:0.00 MB
读字符数:0.89 MB
写字符数:0.01 MB

💡 区分 read_bytesread_charsread_bytes 是磁盘 I/O 的字节数(物理层面),read_chars 是进程通过系统调用读到的字符数(逻辑层面)。对于运维监控,read_bytes / write_bytes 更直接反映磁盘负载。

3.6 进程线程信息
import psutil

p = psutil.Process(1)

# 线程数量
print(f"线程数:{p.num_threads()}")

# 线程详细信息
threads = p.threads()
for thread in threads:
    print(f"  线程 ID: {thread.id}, 用户态: {thread.user_time:.2f}s, 系统态: {thread.system_time:.2f}s")

运行结果

线程数:1
  线程 ID: 1, 用户态: 0.03s, 系统态: 0.04s

四、进程资源监控实战

4.1 获取所有进程的 CPU 和内存使用情况
import psutil

def get_all_processes_info():
    """获取所有进程的 PID、名称、CPU% 和 内存%"""
    processes = []

    for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
        try:
            processes.append({
                'pid': proc.info['pid'],
                'name': proc.info['name'],
                'cpu_percent': proc.info['cpu_percent'] or 0.0,
                'memory_percent': proc.info['memory_percent'] or 0.0,
            })
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    return processes

# 获取所有进程信息
all_procs = get_all_processes_info()
print(f"共 {len(all_procs)} 个进程")

# 按 CPU 使用率排序
sorted_by_cpu = sorted(all_procs, key=lambda x: x['cpu_percent'], reverse=True)
print("\nCPU 使用率 Top 5:")
for p in sorted_by_cpu[:5]:
    print(f"  PID {p['pid']:>6} {p['name']:20} CPU: {p['cpu_percent']:>6.2f}%  MEM: {p['memory_percent']:>6.2f}%")

运行结果

共 145 个进程

CPU 使用率 Top 5:
  PID      1 systemd                CPU:   0.00%  MEM:   0.33%
  PID      2 kthreadd               CPU:   0.00%  MEM:   0.00%
  PID      3 rcu_gp                 CPU:   0.00%  MEM:   0.00%
  PID      4 rcu_par_gp             CPU:   0.00%  MEM:   0.00%
  PID      6 kworker/0:0H-events_   CPU:   0.00%  MEM:   0.00%
4.2 监控进程状态变化
import psutil
import time

def monitor_process(pid, duration=10):
    """监控指定进程的资源使用情况"""
    try:
        p = psutil.Process(pid)
        print(f"开始监控进程 {p.name()} (PID: {pid}),持续 {duration} 秒")
        print("-" * 60)

        for i in range(duration):
            cpu = p.cpu_percent(interval=1)
            mem = p.memory_percent()
            mem_rss = p.memory_info().rss / (1024**2)
            status = p.status()

            print(f"第 {i+1:2} 秒 | CPU: {cpu:5.1f}% | MEM: {mem:5.1f}% | RSS: {mem_rss:6.1f}MB | 状态: {status}")
    except psutil.NoSuchProcess:
        print(f"进程 {pid} 不存在或已终止")
    except psutil.AccessDenied:
        print(f"无权限访问进程 {pid}")
4.3 检测僵尸进程
import psutil

def find_zombie_processes():
    """查找系统中的僵尸进程"""
    zombies = []

    for proc in psutil.process_iter(['pid', 'name', 'status', 'ppid']):
        try:
            if proc.info['status'] == 'zombie':
                zombies.append({
                    'pid': proc.info['pid'],
                    'name': proc.info['name'],
                    'ppid': proc.info['ppid'],
                    'status': proc.info['status']
                })
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    return zombies

zombies = find_zombie_processes()
if zombies:
    print(f"⚠️ 发现 {len(zombies)} 个僵尸进程:")
    for z in zombies:
        print(f"  PID: {z['pid']}, 名称: {z['name']}, 父进程 PID: {z['ppid']}")
else:
    print("✅ 未发现僵尸进程")

运行结果

✅ 未发现僵尸进程

五、子进程管理——subprocess.Popen

除了监控已有的进程,运维脚本还需要主动执行系统命令并获取输出。subprocess.Popen 是 Python 中创建和管理子进程的核心工具。

5.1 为什么不用 os.system()
对比项 os.system() subprocess.Popen
获取输出 ❌ 无法获取命令输出 ✅ 可通过 stdout 管道获取
错误处理 ❌ 只能获取返回值 ✅ 可获取 stderr 分离处理
安全性 ⚠️ 命令注入风险 ✅ 可传参列表避免注入
控制粒度 ❌ 只能等待执行结束 ✅ 可交互、可超时、可终止
5.2 Popen 基本用法
import subprocess

# 方式一:传参列表(推荐,更安全)
process = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

# 方式二:传字符串(需设置 shell=True,有命令注入风险)
# process = subprocess.Popen('ls -l', shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

# 与进程交互,获取输出和错误
output, error = process.communicate()

# 检查执行结果
if process.returncode == 0:
    print("命令执行成功!")
    print(output.decode('utf-8'))
else:
    print(f"命令执行失败,返回码:{process.returncode}")
    print(error.decode('utf-8'))

运行结果

命令执行成功!
total 52
-rw-r--r-- 1 root root  439 Feb  3 12:14 cpu.py
-rw-r--r-- 1 root root  298 Feb  3 13:25 disk.py
-rw-r--r-- 1 root root  201 Feb  3 13:04 mem.py
-rw-r--r-- 1 root root  147 Feb  3 13:37 network.py
...
5.3 Popen 核心参数
参数 含义 常用值
stdout 标准输出重定向 subprocess.PIPE(捕获输出)
stderr 标准错误重定向 subprocess.PIPE(捕获错误)或 subprocess.STDOUT(合并到 stdout)
stdin 标准输入 subprocess.PIPE(向进程输入数据)
cwd 工作目录 字符串路径
env 环境变量 字典
timeout 超时时间(秒) 整数或浮点数
shell 是否通过 shell 执行 True / False(推荐 False
5.4 常用方法
方法 作用 返回值
communicate(input=None) 与进程交互,发送输入并读取输出 (stdout, stderr)
wait(timeout=None) 等待进程结束 返回码
poll() 检查进程是否已结束(非阻塞) None(运行中)或返回码(已结束)
terminate() 终止进程(发送 SIGTERM) -
kill() 强制终止进程(发送 SIGKILL) -
pid 子进程 PID 整数
5.5 实战示例:执行命令并捕获输出
import subprocess
import time

def run_command(cmd, timeout=30):
    """
    执行系统命令并返回输出
    参数:
        cmd: 命令列表(如 ['free', '-m'])
        timeout: 超时时间(秒)
    返回:
        (success, stdout, stderr)
    """
    try:
        process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True  # 自动解码为字符串(Python 3.7+)
        )

        stdout, stderr = process.communicate(timeout=timeout)

        if process.returncode == 0:
            return True, stdout, stderr
        else:
            return False, stdout, stderr

    except subprocess.TimeoutExpired:
        process.kill()
        return False, "", f"命令执行超时({timeout}秒)"
    except Exception as e:
        return False, "", str(e)

# 测试:获取内存信息
success, stdout, stderr = run_command(['free', '-m'])
if success:
    print("内存信息:")
    print(stdout)
else:
    print(f"执行失败:{stderr}")

运行结果

内存信息:
              total        used        free      shared  buff/cache   available
Mem:           3869         534         587          25        2748        3026
Swap:          8191           0        8191

六、综合实战:进程资源 Top N 监控脚本

需求描述:编写一个 Python 脚本,定期采集系统中 CPU 使用率和内存使用率最高的 Top N 进程,并输出到控制台或日志文件,用于快速定位资源消耗大户。

设计思路

  • 使用 process_iter() 一次性获取所有进程的 CPU 和内存信息
  • 分别按 CPU 和内存排序,取 Top N
  • 支持命令行参数指定 N 值和输出格式
  • 增加异常处理,避免单个进程读取失败影响整体
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
进程资源 Top N 监控脚本
用途:采集系统中 CPU 和内存使用率最高的 Top N 进程
作者:Whisky
"""

import psutil
import time
import argparse
from datetime import datetime

def get_top_processes(top_n=5, interval=1):
    """
    获取 CPU 和内存使用率 Top N 的进程
    参数:
        top_n: 取 Top N
        interval: CPU 采样间隔(秒)
    返回:
        (top_cpu, top_mem)
    """
    processes_cpu = []
    processes_mem = []

    # 第一次遍历:获取进程基本信息,为 cpu_percent 预热
    for proc in psutil.process_iter(['pid', 'name']):
        try:
            proc.cpu_percent(interval=0)  # 预热
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    # 等待采样间隔(让 cpu_percent 有实际数据)
    time.sleep(interval)

    # 第二次遍历:采集实际数据
    for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent', 'status']):
        try:
            info = proc.info
            pid = info['pid']
            name = info['name']
            cpu = info['cpu_percent'] or 0.0
            mem = info['memory_percent'] or 0.0
            status = info['status']

            proc_info = {
                'pid': pid,
                'name': name,
                'cpu': cpu,
                'mem': mem,
                'status': status
            }
            processes_cpu.append(proc_info)
            processes_mem.append(proc_info)

        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    # 分别按 CPU 和内存排序
    top_cpu = sorted(processes_cpu, key=lambda x: x['cpu'], reverse=True)[:top_n]
    top_mem = sorted(processes_mem, key=lambda x: x['mem'], reverse=True)[:top_n]

    return top_cpu, top_mem

def format_table(title, headers, rows):
    """格式化输出表格"""
    print("\n" + "=" * 70)
    print(f"  {title}")
    print("=" * 70)

    # 计算每列宽度
    col_widths = [len(h) for h in headers]
    for row in rows:
        for i, col in enumerate(row):
            col_widths[i] = max(col_widths[i], len(str(col)))

    # 打印表头
    header_line = " | ".join(h.ljust(col_widths[i]) for i, h in enumerate(headers))
    print(header_line)
    print("-" * len(header_line))

    # 打印数据行
    for row in rows:
        line = " | ".join(str(col).ljust(col_widths[i]) for i, col in enumerate(row))
        print(line)

    print("=" * 70)

def main():
    parser = argparse.ArgumentParser(description='进程资源 Top N 监控工具')
    parser.add_argument('-n', '--top', type=int, default=5, help='显示 Top N 个进程(默认 5)')
    parser.add_argument('-i', '--interval', type=float, default=1, help='CPU 采样间隔秒数(默认 1)')
    args = parser.parse_args()

    print("=" * 70)
    print(f"  进程资源 Top {args.top} 监控")
    print(f"  采集时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"  采样间隔:{args.interval} 秒")
    print("=" * 70)

    try:
        top_cpu, top_mem = get_top_processes(top_n=args.top, interval=args.interval)

        # 输出 CPU Top N
        cpu_rows = []
        for p in top_cpu:
            cpu_rows.append([
                p['pid'],
                p['name'][:20],
                f"{p['cpu']:.2f}%",
                f"{p['mem']:.2f}%",
                p['status']
            ])
        format_table(
            f"🔥 CPU 使用率 Top {args.top}",
            ['PID', '进程名称', 'CPU%', 'MEM%', '状态'],
            cpu_rows
        )

        # 输出内存 Top N
        mem_rows = []
        for p in top_mem:
            mem_rows.append([
                p['pid'],
                p['name'][:20],
                f"{p['mem']:.2f}%",
                f"{p['cpu']:.2f}%",
                p['status']
            ])
        format_table(
            f"💾 内存使用率 Top {args.top}",
            ['PID', '进程名称', 'MEM%', 'CPU%', '状态'],
            mem_rows
        )

    except KeyboardInterrupt:
        print("\n\n用户中断,退出监控")
    except Exception as e:
        print(f"\n❌ 监控过程出现异常:{e}")

    print("\n" + "=" * 70)
    print("  采集完成!")
    print("=" * 70)

if __name__ == "__main__":
    main()

运行效果

======================================================================
  进程资源 Top 5 监控
  采集时间:2026-07-19 10:23:45
  采样间隔:1.0 秒
======================================================================

======================================================================
  🔥 CPU 使用率 Top 5
======================================================================
PID  | 进程名称               | CPU%   | MEM%   | 状态
----------------------------------------------------------------------
1234 | python3                | 45.23% | 12.34% | running
5678 | mysql                  | 23.45% | 45.67% | sleeping
9012 | nginx                  | 12.34% | 3.45%  | running
3456 | redis-server           | 8.90%  | 8.90%  | sleeping
7890 | java                   | 5.67%  | 23.45% | running
======================================================================

======================================================================
  💾 内存使用率 Top 5
======================================================================
PID  | 进程名称               | MEM%   | CPU%   | 状态
----------------------------------------------------------------------
5678 | mysql                  | 45.67% | 23.45% | sleeping
7890 | java                   | 23.45% | 5.67%  | running
1234 | python3                | 12.34% | 45.23% | running
3456 | redis-server           | 8.90%  | 8.90%  | sleeping
9012 | nginx                  | 3.45%  | 12.34% | running
======================================================================

======================================================================
  采集完成!
======================================================================

核心模块解读

函数 功能 核心逻辑
get_top_processes() 采集所有进程并按 CPU/内存排序 process_iter() + sorted()
format_table() 格式化表格输出 动态计算列宽,对齐输出
main() 解析命令行参数,调用采集和输出 argparse + 异常处理

📝 总结与知识点一览表

核心 API 速查表

功能 核心 API 返回类型 关键字段/属性
所有进程 PID psutil.pids() list[int] -
遍历所有进程 psutil.process_iter(attrs) 迭代器 proc.info
进程基本信息 p.name(), p.pid, p.ppid(), p.status(), p.create_time() 各类 -
进程路径 p.exe(), p.cwd(), p.cmdline() str / list[str] -
进程 CPU p.cpu_times(), p.cpu_percent() 命名元组 / float user, system
进程内存 p.memory_info(), p.memory_percent() 命名元组 / float rss, vms, shared
进程 I/O p.io_counters() 命名元组 read_bytes, write_bytes
进程线程 p.num_threads(), p.threads() int / list[Thread] id, user_time
进程状态 p.status() str running, sleeping, zombie
子进程管理 subprocess.Popen() Popen 对象 stdout, stderr, returncode

运维命令对照表

运维需求 Shell 命令 psutil 替代方案
查看所有进程 ps -ef psutil.process_iter()
查看进程 CPU top -p <PID> p.cpu_percent()
查看进程内存 pmap <PID> p.memory_info()
查看进程打开文件 lsof -p <PID> p.open_files()(Linux)
查看进程网络连接 `netstat -anp grep `
查看进程树 pstree -p <PID> p.parent(), p.children()
查看进程环境变量 cat /proc/<PID>/environ p.environ()(需权限)

常见错误排查指南

错误现象 大概率原因 解决方案
psutil.NoSuchProcess 进程在采集过程中终止 使用 try-except 捕获,继续下一个
psutil.AccessDenied 无权限访问该进程(如内核进程) 使用 try-except 捕获,跳过该进程
cpu_percent() 返回 0.0 首次调用没有采样间隔 先调用一次 cpu_percent(0) 预热,再 sleep(interval) 后再次调用
subprocess.TimeoutExpired 命令执行时间超过超时阈值 适当调整 timeout 参数,或检查命令是否卡死
僵尸进程数量持续增长 父进程未正确回收子进程 检查父进程代码,确保调用 wait() 或使用 subprocess 管理子进程

学习/生产最佳实践

  1. 遍历进程时必须处理异常:进程可能在遍历过程中结束,或某些内核进程无权限访问。务必用 try-except (psutil.NoSuchProcess, psutil.AccessDenied) 包裹,确保脚本不会因单个进程失败而中断。
  2. cpu_percent() 需要预热:进程对象的 cpu_percent() 首次调用返回 0.0,需要先调用一次(或传 interval=0)进行预热,等待间隔后再获取有效值。更推荐使用 process_iter()attrs=['cpu_percent'] 一次性获取。
  3. 优先使用 process_iter() 而非 pids()process_iter() 支持一次性获取多个属性,减少系统调用次数,性能更好。
  4. subprocess 优先传参列表:使用 ['ls', '-l'] 而非 'ls -l',避免 shell 注入风险,也无需设置 shell=True
  5. 监控数据需持久化:生产环境中,进程监控数据应定期采集并写入日志或时序数据库(如 InfluxDB、Prometheus),用于后续分析和告警。

🔜 下期预告

本期我们完成了 psutil 模块的进程级监控学习——从获取进程列表、查看进程详细信息(CPU/内存/IO/线程)、到利用 subprocess.Popen 执行系统命令并捕获输出,最后通过“进程资源 Top N 监控脚本”将各知识点整合为实战工具。

至此,psutil 模块的前后两篇已全部完结:

  • 上·前篇(第10期):系统基础监控——CPU/内存/磁盘/网络
  • 上·后篇(第11期):进程级监控与子进程管理——psutil进阶

如果您在使用过程中发现任何问题或有改进建议,欢迎随时提出!

下一期 ,我们将进入 IP 地址规划 专题,学习 IPy 模块的使用——从 IP 地址的基础处理、网段计算、地址转换,到网段比较与重叠判断,全面掌握 Python 中的网络地址规划能力。敬请期待!


— Compiled and Authored by Whisky — July 19 th, 2026
Logo

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐