Python系统监控利器：psutil详解

引言

psutil(Python system and process utilities)是一个跨平台的Python库，它允许开发人员轻松获取系统运行时的各种信息。这个库最初由Giampaolo Rodola在2009年创建，现在已经成为Python生态系统中系统监控领域的标准工具之一。psutil提供了一种简单而统一的方式来获取系统利用率信息(CPU、内存、磁盘、网络)和进程管理功能，不需要依赖系统特定的命令行工具。

在系统监控和资源管理领域，psutil扮演着至关重要的角色。它可以帮助开发人员：

• 实时监控服务器资源使用情况
• 诊断系统性能瓶颈
• 自动管理进程生命周期
• 构建自定义的系统监控工具
• 实现自动化运维任务

安装与基本配置

安装方法

psutil可以通过Python的包管理工具pip轻松安装：

pip install psutil

对于特定版本的需求，可以指定版本号安装：

pip install psutil==5.9.0

兼容性说明

psutil具有出色的跨平台兼容性，支持以下操作系统：

• Windows (32位和64位)
• Linux (各主流发行版)
• macOS
• FreeBSD, OpenBSD, NetBSD
• Sun Solaris
• AIX

支持的Python版本包括：

• Python 2.7 (psutil 5.x是最后一个支持Python 2的版本)
• Python 3.4+
• PyPy

核心功能与用法

获取系统信息

CPU使用率监控

psutil提供了多种CPU监控功能：

import psutil

# 获取CPU逻辑核心数
print(f"逻辑CPU核心数: {psutil.cpu_count()}")

# 获取CPU物理核心数(不包括超线程)
print(f"物理CPU核心数: {psutil.cpu_count(logical=False)}")

# 获取CPU使用率(阻塞1秒)
print(f"CPU使用率: {psutil.cpu_percent(interval=1)}%")

# 获取每个CPU核心的使用率
print(f"各核心使用率: {psutil.cpu_percent(interval=1, percpu=True)}")

# 获取CPU频率信息
cpu_freq = psutil.cpu_freq()
print(f"当前频率: {cpu_freq.current}MHz")
print(f"最小频率: {cpu_freq.min}MHz")
print(f"最大频率: {cpu_freq.max}MHz")

# 获取CPU统计信息(上下文切换、中断等)
print(f"CPU统计: {psutil.cpu_stats()}")

内存使用情况

psutil可以详细监控系统内存使用情况：

# 获取物理内存信息
mem = psutil.virtual_memory()
print(f"总内存: {mem.total/1024/1024:.2f} MB")
print(f"可用内存: {mem.available/1024/1024:.2f} MB")
print(f"已用内存: {mem.used/1024/1024:.2f} MB")
print(f"内存使用率: {mem.percent}%")

# 获取交换内存信息
swap = psutil.swap_memory()
print(f"交换内存总量: {swap.total/1024/1024:.2f} MB")
print(f"已用交换内存: {swap.used/1024/1024:.2f} MB")
print(f"交换内存使用率: {swap.percent}%")

磁盘I/O和分区信息

磁盘监控功能示例：

# 获取磁盘分区信息
partitions = psutil.disk_partitions()
for partition in partitions:
print(f"设备: {partition.device}")
print(f"挂载点: {partition.mountpoint}")
print(f"文件系统类型: {partition.fstype}")

# 获取磁盘使用情况
usage = psutil.disk_usage(partition.mountpoint)
print(f"总空间: {usage.total/1024/1024:.2f} MB")
print(f"已用空间: {usage.used/1024/1024:.2f} MB")
print(f"可用空间: {usage.free/1024/1024:.2f} MB")
print(f"使用率: {usage.percent}%")

# 获取磁盘IO统计
disk_io = psutil.disk_io_counters()
print(f"读取次数: {disk_io.read_count}")
print(f"写入次数: {disk_io.write_count}")
print(f"读取字节数: {disk_io.read_bytes/1024/1024:.2f} MB")
print(f"写入字节数: {disk_io.write_bytes/1024/1024:.2f} MB")

网络接口统计

网络监控功能：

# 获取网络接口信息
net_if = psutil.net_if_addrs()
for interface, addrs in net_if.items():
print(f"接口: {interface}")
for addr in addrs:
print(f" IP地址: {addr.address}")
print(f" 网络掩码: {addr.netmask}")
print(f" 广播地址: {addr.broadcast}")

# 获取网络IO统计
net_io = psutil.net_io_counters()
print(f"发送字节数: {net_io.bytes_sent/1024:.2f} KB")
print(f"接收字节数: {net_io.bytes_recv/1024:.2f} KB")
print(f"发送数据包数: {net_io.packets_sent}")
print(f"接收数据包数: {net_io.packets_recv}")

# 获取当前网络连接
connections = psutil.net_connections()
for conn in connections:
print(f"协议: {conn.type}")
print(f"本地地址: {conn.laddr}")
print(f"远程地址: {conn.raddr}")
print(f"状态: {conn.status}")

进程管理

列出当前运行的所有进程

# 获取所有进程ID
pids = psutil.pids()

# 遍历所有进程
for proc in psutil.process_iter(['pid', 'name', 'username']):
try:
# 获取进程详情
pinfo = proc.as_dict(attrs=['pid', 'name', 'cpu_percent', 'memory_percent'])
print(pinfo)
except psutil.NoSuchProcess:
pass

获取特定进程的详细信息

# 通过PID获取进程对象
pid = 1234 # 替换为实际PID
try:
p = psutil.Process(pid)

# 获取进程基本信息
print(f"进程名: {p.name()}")
print(f"执行路径: {p.exe()}")
print(f"工作目录: {p.cwd()}")
print(f"启动命令: {p.cmdline()}")
print(f"启动时间: {p.create_time()}")
print(f"用户名: {p.username()}")

# 获取资源使用情况
print(f"CPU使用率: {p.cpu_percent(interval=1.0)}%")
print(f"内存使用率: {p.memory_percent()}%")
mem_info = p.memory_info()
print(f"常驻内存: {mem_info.rss/1024/1024:.2f} MB")
print(f"虚拟内存: {mem_info.vms/1024/1024:.2f} MB")

# 获取线程信息
print(f"线程数: {p.num_threads()}")

# 获取打开的文件
print(f"打开的文件: {p.open_files()}")

# 获取网络连接
print(f"网络连接: {p.connections()}")

except psutil.NoSuchProcess:
print(f"进程 {pid} 不存在")

进程的启动、终止与管理

# 启动新进程
import subprocess
proc = subprocess.Popen(["python", "script.py"])
psutil_proc = psutil.Process(proc.pid)

# 终止进程
psutil_proc.terminate() # 优雅终止
psutil_proc.kill() # 强制终止

# 挂起和恢复进程
psutil_proc.suspend()
psutil_proc.resume()

# 等待进程结束
psutil_proc.wait(timeout=10)

系统性能优化与监控

实时监控系统资源

import time

def monitor_system(interval=1):
"""实时监控系统资源"""
while True:
# CPU使用率
cpu_percent = psutil.cpu_percent(interval=interval, percpu=True)

# 内存使用情况
mem = psutil.virtual_memory()

# 磁盘使用情况
disk = psutil.disk_usage('/')

# 网络IO
net_io = psutil.net_io_counters()

# 打印监控信息
print("\\n" + "="*50)
print(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"CPU使用率: {cpu_percent}")
print(f"内存使用: {mem.percent}%")
print(f"磁盘使用: {disk.percent}%")
print(f"网络: 上传 {net_io.bytes_sent/1024:.2f}KB, 下载 {net_io.bytes_recv/1024:.2f}KB")

time.sleep(interval)

# monitor_system()

生成资源使用报告

def generate_report(duration=60, interval=5):
"""生成系统资源使用报告"""
records = []
end_time = time.time() + duration

while time.time() < end_time:
record = {
'timestamp': time.time(),
'cpu': psutil.cpu_percent(interval=interval),
'memory': psutil.virtual_memory().percent,
'disk': psutil.disk_usage('/').percent,
'network_sent': psutil.net_io_counters().bytes_sent,
'network_recv': psutil.net_io_counters().bytes_recv
}
records.append(record)

# 计算统计数据
cpu_avg = sum(r['cpu'] for r in records) / len(records)
mem_avg = sum(r['memory'] for r in records) / len(records)

report = {
'duration': duration,
'cpu_avg': cpu_avg,
'mem_avg': mem_avg,
'peak_cpu': max(r['cpu'] for r in records),
'peak_mem': max(r['memory'] for r in records),
'network_total': (records[-1]['network_sent'] – records[0]['network_sent'] +
records[-1]['network_recv'] – records[0]['network_recv']) / 1024
}

return report

# print(generate_report())

结合日志分析长期趋势

import logging
from logging.handlers import TimedRotatingFileHandler

def setup_monitoring_logger():
"""设置监控日志"""
logger = logging.getLogger('system_monitor')
logger.setLevel(logging.INFO)

# 每天轮换日志，保留7天
handler = TimedRotatingFileHandler(
'system_monitor.log',
when='D',
interval=1,
backupCount=7
)
formatter = logging.Formatter('%(asctime)s – %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

return logger

def log_system_stats(logger, interval=300):
"""定期记录系统状态到日志"""
while True:
stats = {
'cpu': psutil.cpu_percent(interval=1),
'mem': psutil.virtual_memory().percent,
'disk': psutil.disk_usage('/').percent
}
logger.info(f"CPU: {stats['cpu']}% | MEM: {stats['mem']}% | DISK: {stats['disk']}%")
time.sleep(interval)

# logger = setup_monitoring_logger()
# log_system_stats(logger)

高级应用与技巧

多平台适配的注意事项

在使用psutil进行跨平台开发时，需要注意以下差异：

结合其他工具进行数据可视化

使用Pandas分析数据

import pandas as pd

def collect_system_data(samples=10, interval=1):
"""收集系统数据并返回DataFrame"""
data = []
for _ in range(samples):
record = {
'timestamp': pd.Timestamp.now(),
'cpu': psutil.cpu_percent(interval=interval),
'memory': psutil.virtual_memory().percent,
'disk': psutil.disk_usage('/').percent
}
data.append(record)
time.sleep(interval)

return pd.DataFrame(data)

# df = collect_system_data()
# print(df.describe())

使用Matplotlib绘制图表

import matplotlib.pyplot as plt

def plot_system_metrics(df):
"""绘制系统指标图表"""
fig, axes = plt.subplots(nrows=3, ncols=1, figsize=(10, 8))

# CPU使用率
df.plot(x='timestamp', y='cpu', ax=axes[0], title='CPU Usage (%)', legend=False)
axes[0].set_ylim(0, 100)

# 内存使用率
df.plot(x='timestamp', y='memory', ax=axes[1], title='Memory Usage (%)', legend=False)
axes[1].set_ylim(0, 100)

# 磁盘使用率
df.plot(x='timestamp', y='disk', ax=axes[2], title='Disk Usage (%)', legend=False)
axes[2].set_ylim(0, 100)

plt.tight_layout()
plt.show()

# plot_system_metrics(df)

异常处理与性能调优

健壮的进程监控

def safe_process_info(pid):
"""安全获取进程信息，处理各种异常"""
try:
p = psutil.Process(pid)
return {
'pid': pid,
'name': p.name(),
'status': p.status(),
'cpu': p.cpu_percent(interval=0.1),
'memory': p.memory_percent(),
'threads': p.num_threads()
}
except psutil.NoSuchProcess:
return {'error': 'Process not found'}
except psutil.AccessDenied:
return {'error': 'Access denied'}
except psutil.ZombieProcess:
return {'error': 'Zombie process'}
except Exception as e:
return {'error': str(e)}

性能优化技巧

实际案例

服务器监控脚本的实现

#!/usr/bin/env python3
"""
服务器资源监控脚本
监控CPU、内存、磁盘、网络使用情况
当资源超过阈值时发送告警
"""

import psutil
import time
import smtplib
from email.mime.text import MIMEText

# 配置参数
THRESHOLDS = {
'cpu': 90, # CPU使用率阈值(%)
'memory': 90, # 内存使用率阈值(%)
'disk': 90, # 磁盘使用率阈值(%)
}

ALERT_INTERVAL = 3600 # 告警间隔(秒)
MONITOR_INTERVAL = 5 # 监控间隔(秒)

# 邮件配置
SMTP_SERVER = 'smtp.example.com'
SMTP_PORT = 587
EMAIL_USER = 'monitor@example.com'
EMAIL_PASS = 'password'
EMAIL_TO = 'admin@example.com'

last_alert_time = 0

def check_thresholds():
"""检查各项指标是否超过阈值"""
alerts = []

# CPU检查
cpu_percent = psutil.cpu_percent(interval=1)
if cpu_percent > THRESHOLDS['cpu']:
alerts.append(f"CPU使用率过高: {cpu_percent}%")

# 内存检查
mem_percent = psutil.virtual_memory().percent
if mem_percent > THRESHOLDS['memory']:
alerts.append(f"内存使用率过高: {mem_percent}%")

# 磁盘检查
disk_percent = psutil.disk_usage('/').percent
if disk_percent > THRESHOLDS['disk']:
alerts.append(f"磁盘使用率过高: {disk_percent}%")

return alerts

def send_alert(alerts):
"""发送告警邮件"""
global last_alert_time

current_time = time.time()
if current_time – last_alert_time < ALERT_INTERVAL:
return

subject = "服务器资源告警"
body = "\\n".join(alerts)

msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = EMAIL_USER
msg['To'] = EMAIL_TO

try:
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
server.starttls()
server.login(EMAIL_USER, EMAIL_PASS)
server.send_message(msg)
last_alert_time = current_time
print(f"告警已发送: {alerts}")
except Exception as e:
print(f"发送告警失败: {e}")

def main():
"""主监控循环"""
print("启动服务器监控…")
while True:
alerts = check_thresholds()
if alerts:
send_alert(alerts)
time.sleep(MONITOR_INTERVAL)

if __name__ == '__main__':
main()

自动化运维任务示例

#!/usr/bin/env python3
"""
自动化运维脚本
1. 监控指定进程，如果崩溃则自动重启
2. 清理过期的日志文件
3. 定期备份重要数据
"""

import psutil
import os
import time
import shutil
import subprocess
from datetime import datetime, timedelta

# 配置参数
PROCESS_NAME = "my_service" # 要监控的进程名
PROCESS_CMD = ["python", "service.py"] # 进程启动命令
LOG_DIR = "/var/log/myapp" # 日志目录
BACKUP_DIR = "/backups" # 备份目录
DATA_DIR = "/data" # 要备份的数据目录
RETENTION_DAYS = 7 # 日志和备份保留天数
CHECK_INTERVAL = 60 # 检查间隔(秒)

def find_process():
"""查找指定进程"""
for proc in psutil.process_iter(['name', 'cmdline']):
try:
if proc.info['name'] == PROCESS_NAME or \\
(proc.info['cmdline'] and proc.info['cmdline'][0] == PROCESS_CMD[0]):
return proc
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return None

def start_process():
"""启动进程"""
print(f"启动进程: {' '.join(PROCESS_CMD)}")
subprocess.Popen(PROCESS_CMD)

def cleanup_files(directory, pattern, days):
"""清理过期文件"""
now = datetime.now()
cutoff = now – timedelta(days=days)

for filename in os.listdir(directory):
if filename.endswith(pattern):
filepath = os.path.join(directory, filename)
mtime = datetime.fromtimestamp(os.path.getmtime(filepath))
if mtime < cutoff:
print(f"删除过期文件: {filepath}")
os.remove(filepath)

def create_backup():
"""创建数据备份"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_name = f"backup_{timestamp}.tar.gz"
backup_path = os.path.join(BACKUP_DIR, backup_name)

print(f"创建备份: {backup_path}")
shutil.make_archive(backup_path.replace('.tar.gz', ''), 'gztar', DATA_DIR)
return backup_path

def main():
"""主循环"""
print("启动自动化运维脚本…")

while True:
# 检查进程
proc = find_process()
if proc is None:
print("进程未运行，尝试启动…")
start_process()
else:
print(f"进程运行中，PID: {proc.pid}")

# 清理过期日志
print("清理过期日志…")
cleanup_files(LOG_DIR, ".log", RETENTION_DAYS)

# 创建定期备份
if datetime.now().hour == 2: # 每天凌晨2点备份
print("执行每日备份…")
create_backup()
print("清理过期备份…")
cleanup_files(BACKUP_DIR, ".tar.gz", RETENTION_DAYS)

time.sleep(CHECK_INTERVAL)

if __name__ == '__main__':
main()

常见问题与解决方案

常见错误与调试方法

性能瓶颈分析

总结

psutil的优势与适用场景

psutil作为Python系统监控的标准库，具有以下显著优势：

psutil非常适合以下场景：

• 服务器监控和告警系统
• 自动化运维工具开发
• 系统性能分析和调优
• 资源使用统计和报告生成
• 进程管理和监控工具

未来发展方向与社区支持

psutil项目持续活跃发展，未来可能的方向包括：

psutil拥有一个活跃的开源社区，用户可以通过以下方式获取支持和参与贡献：

• GitHub仓库: https://github.com/giampaolo/psutil
• 官方文档: https://psutil.readthedocs.io/
• 邮件列表和问题追踪

通过psutil，Python开发者可以轻松构建强大的系统监控和管理工具，极大地简化了与系统资源交互的复杂性，是每个系统管理员和DevOps工程师工具箱中不可或缺的利器。