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%

内存字段说明

字段全称含义运维意义
RSSResident Set Size物理内存中实际占用的内存页最重要的指标,反映进程实际占用的物理内存
VMSVirtual Memory Size虚拟内存总量(含未分配的地址空间)反映进程申请的总虚拟地址空间
sharedShared Memory与其他进程共享的内存共享库占用的内存
textText Segment可执行代码段大小进程代码的大小
dataData Segment数据段大小进程数据(堆/栈)的大小
dirtyDirty 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返回类型关键字段/属性
所有进程 PIDpsutil.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]-
进程 CPUp.cpu_times(), p.cpu_percent()命名元组 / floatuser, system
进程内存p.memory_info(), p.memory_percent()命名元组 / floatrss, vms, shared
进程 I/Op.io_counters()命名元组read_bytes, write_bytes
进程线程p.num_threads(), p.threads()int / list[Thread]id, user_time
进程状态p.status()strrunning, sleeping, zombie
子进程管理subprocess.Popen()Popen 对象stdout, stderr, returncode

运维命令对照表

运维需求Shell 命令psutil 替代方案
查看所有进程ps -efpsutil.process_iter()
查看进程 CPUtop -p <PID>p.cpu_percent()
查看进程内存pmap <PID>p.memory_info()
查看进程打开文件lsof -p <PID>p.open_files()(Linux)
查看进程网络连接`netstat -anpgrep `
查看进程树pstree -p <PID>p.parent(), p.children()
查看进程环境变量cat /proc/<PID>/environp.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 等多样性计算架构

更多推荐