当前位置：首页 > news >正文

用 Python 实现系统监控与资源管理：深入解析 `psutil` 库

news 2025/5/17 13:13:46

在系统运维、资源监控、性能分析、服务状态诊断等领域，开发者常常需要获取操作系统底层信息，如 CPU 使用率、内存占用、磁盘读写、网络传输、进程详情等。虽然这些功能可以通过 shell 命令实现，如 top、df、ps、iostat 等，但在自动化脚本和跨平台开发中，使用 Python 脚本来完成这些任务更具灵活性和可移植性。

这正是 psutil 的用武之地 —— 一个功能强大的跨平台系统监控库，可以轻松获取各种系统资源状态和进程信息，且支持 Linux、Windows、macOS 等平台。

本文将深入讲解 psutil 的各项功能，并结合实际代码示例，展示如何用 Python 构建自己的系统监控工具、资源分析器、进程管理脚本等。

一、psutil 简介

psutil 是“process and system utilities”的缩写，它是一个 跨平台的系统监控库，用于：

获取 CPU、内存、磁盘、网络 等系统使用情况
管理和获取 进程信息
实现任务监控、资源追踪和系统健康检查
提供类 Unix ps, top, df, netstat, lsof 等命令的功能接口

它的设计初衷是替代用 Python 封装 shell 命令的繁琐做法，提供一致、优雅、性能优异的 API。

二、安装与基本使用

安装方式

使用 pip 安装非常简单：

pip install psutil

安装完成后，几乎可以立即开始使用，无需额外配置。

快速示例：获取当前 CPU 使用率

import psutilprint("当前 CPU 使用率：", psutil.cpu_percent(interval=1), "%")

cpu_percent() 表示在指定时间间隔内（如 1 秒）测量 CPU 总体使用率。

三、CPU 信息监控

1. 获取 CPU 逻辑/物理核心数

print("物理核心数：", psutil.cpu_count(logical=False))
print("逻辑核心数：", psutil.cpu_count(logical=True))

2. 每个核心的使用率

import timefor _ in range(5):usage = psutil.cpu_percent(interval=1, percpu=True)print("每个核心使用率：", usage)

3. 获取 CPU 时间信息

times = psutil.cpu_times()
print(f"user: {times.user}, system: {times.system}, idle: {times.idle}")

这个 API 提供了系统级的 CPU 时间分布。

四、内存信息监控

1. 获取虚拟内存使用情况

mem = psutil.virtual_memory()
print(f"总内存: {mem.total / 1024 ** 3:.2f} GB")
print(f"已使用: {mem.used / 1024 ** 3:.2f} GB")
print(f"空闲: {mem.available / 1024 ** 3:.2f} GB")
print(f"使用率: {mem.percent}%")

2. 获取 swap 交换分区信息

swap = psutil.swap_memory()
print(f"Swap 总量: {swap.total / 1024 ** 3:.2f} GB")
print(f"使用率: {swap.percent}%")

五、磁盘信息监控

1. 获取磁盘分区信息

partitions = psutil.disk_partitions()
for p in partitions:print(f"挂载点: {p.mountpoint}, 类型: {p.fstype}")

2. 获取磁盘使用情况

usage = psutil.disk_usage('/')
print(f"总空间: {usage.total / 1024 ** 3:.2f} GB")
print(f"已用: {usage.used / 1024 ** 3:.2f} GB")
print(f"剩余: {usage.free / 1024 ** 3:.2f} GB")

3. 实时磁盘 IO 情况

io = psutil.disk_io_counters()
print(f"读字节: {io.read_bytes / 1024 ** 2:.2f} MB")
print(f"写字节: {io.write_bytes / 1024 ** 2:.2f} MB")

六、网络信息监控

1. 获取网络 IO 情况

net = psutil.net_io_counters()
print(f"发送: {net.bytes_sent / 1024 ** 2:.2f} MB")
print(f"接收: {net.bytes_recv / 1024 ** 2:.2f} MB")

2. 获取网卡信息

addrs = psutil.net_if_addrs()
for iface, infos in addrs.items():print(f"\n网卡: {iface}")for info in infos:print(f"  地址: {info.address} ({info.family})")

3. 当前网络连接

conns = psutil.net_connections(kind='inet')
for conn in conns[:5]:print(f"连接: {conn.laddr} -> {conn.raddr}, 状态: {conn.status}")

七、进程管理

1. 枚举当前所有进程

for proc in psutil.process_iter(['pid', 'name', 'username']):print(proc.info)

2. 获取指定进程的详细信息

p = psutil.Process(1)  # PID 1 通常是 init 或 systemd
print(f"名称: {p.name()}")
print(f"路径: {p.exe()}")
print(f"状态: {p.status()}")
print(f"CPU使用率: {p.cpu_percent(interval=1.0)}")
print(f"内存使用: {p.memory_info().rss / 1024 ** 2:.2f} MB")

3. 操作进程（终止、挂起、恢复）

p = psutil.Process(1234)
p.suspend()  # 挂起
p.resume()   # 恢复
p.terminate()  # 终止

注意：需要管理员权限或 root 权限才能操作其他用户的进程。

八、构建一个简单的系统资源仪表板

你可以使用 psutil + rich 或 psutil + flask/streamlit 构建漂亮的图形或终端仪表盘。下面是一个简单的终端监控：

import psutil
import time
import osdef monitor():while True:os.system('cls' if os.name == 'nt' else 'clear')print("=== 实时系统资源监控 ===")print(f"CPU 使用率: {psutil.cpu_percent()}%")mem = psutil.virtual_memory()print(f"内存: {mem.used / 1024 ** 3:.2f} GB / {mem.total / 1024 ** 3:.2f} GB ({mem.percent}%)")net = psutil.net_io_counters()print(f"网络: 发送 {net.bytes_sent / 1024 ** 2:.2f} MB, 接收 {net.bytes_recv / 1024 ** 2:.2f} MB")time.sleep(1)if __name__ == "__main__":monitor()

可以扩展为 Web 仪表盘，甚至接入 Grafana、Prometheus 进行数据上报。

九、高级用法与实践案例

1. 编写一个自动杀死高 CPU 占用进程的守护脚本

def auto_kill(threshold=80):for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']):try:if proc.info['cpu_percent'] > threshold:print(f"杀死高 CPU 进程: {proc.info}")psutil.Process(proc.info['pid']).terminate()except (psutil.NoSuchProcess, psutil.AccessDenied):continue

2. 跨平台性能分析工具

你可以用 psutil 收集数据并记录日志：

import json
import datetimedef log_stats():stats = {"time": str(datetime.datetime.now()),"cpu": psutil.cpu_percent(),"mem": psutil.virtual_memory().percent,"disk": psutil.disk_usage('/').percent,}with open("syslog.json", "a") as f:f.write(json.dumps(stats) + "\n")