Python 跨平台系统资源监控实践
1. 需求分析
1.1. 需求
- 智能系统使用最新的数据进行在线机器学习模型训练
- 智能系统预测功能使用时序模型，例如prophet模型需要在线学习
- 智能系统优化功能使用优化算法，例如粒子群算法定时在线优化计算
1.2. 智能系统关键资源瓶颈分析
- 在线机器学习训练模块
- 存在持续性的资源竞争问题，模型增量训练过程会动态占用大量CPU和内存资源
- 可能导致实时控制指令延迟和服务响应超时等关键业务中断
- 时序预测（在线学习）模块
- Prophet模型的周期性季节项拟合属于计算密集型操作
- 资源占用呈现明显的峰值特征，可能导致系统瞬时过载
- 智能优化计算模块
- 基于粒子群算法等优化器会定时产生计算负载高峰
- 存在三方面挑战：
- 与预测任务的计算资源竞争
- 迭代收敛时间的不确定性
- 可能引发的级联响应延迟
因此，需要系统资源监控，合理分配资源。
2. Python 跨平台系统资源监控实现
要实现一个不依赖操作系统的Python系统资源监控工具，我们可以使用psutil库，它是一个跨平台库，可以获取系统利用率信息（CPU、内存、磁盘、网络等）。
2.1. psutil 库核心功能
psutil（process and system utilities）是一个跨平台库，用于获取 系统运行状态 和 进程信息，支持 Windows、Linux、macOS 等主流操作系统。
2.1.1. 核心监控能力
| 类别 | 可获取信息 |
|---|---|
| CPU | 使用率（整体/单核）、频率、核心数、时间（用户/系统/空闲） |
| 内存 | 物理内存/交换内存总量、已用、剩余、使用率 |
| 磁盘 | 分区信息、使用率、IO 读写次数/字节数、读写时间 |
| 网络 | 连接信息（TCP/UDP）、收发字节数、包数、错误数 |
| 进程 | 进程列表、CPU/内存占用、打开文件、线程数、命令行参数、环境变量 |
| 传感器 | 温度、风扇转速（部分系统支持） |
| 系统信息 | 启动时间、用户登录信息、主机名 |
2.1.2. 跨平台特性
- 自动适配不同操作系统，API 接口一致
- 无需调用系统命令（如
top、tasklist），纯 Python 实现
2.1.3. psutil 进阶技巧
1. 性能优化
- 使用
oneshot()上下文管理器批量获取进程信息（减少系统调用）：with proc.oneshot():print(proc.name(), proc.cpu_percent(), proc.memory_info())
2. 特殊场景处理
- 容器环境监控：通过
/proc文件系统直接读取 cgroups 数据 - 短生命周期进程：用
cpu_times()差值计算瞬时 CPU 使用率
2.1.4. 同类工具对比
| 工具 | 优势 | 劣势 |
|---|---|---|
psutil | 跨平台、API 简洁、功能全面 | 无原生告警功能 |
os 模块 | Python 内置 | 功能有限 |
subprocess | 可调用系统命令 | 需要解析文本输出 |
| Prometheus | 适合分布式监控 | 需要额外组件部署 |
2.1.5. 典型应用场景
- 运维监控：服务器健康状态实时监控
- 性能分析：定位 Python 应用的内存泄漏/CPU 瓶颈
- 安全审计：检测异常进程（如挖矿病毒）
- 资源限制：实现进程资源配额管理
- 自动化测试：记录测试过程中的系统负载
通过 psutil，开发者可以用 Python 快速构建从简单脚本到企业级监控系统的各种工具，其设计平衡了易用性和功能性，是系统监控领域的首选库。
2.2. 安装依赖
首先需要安装psutil库：
pip install psutil
2.3. 功能及完整实践代码
2.3.1. 功能
- 监控cpu、内存、硬盘
- 监控应用
- 监控数据写入csv
- 记录监控日志
监控文件存储结构：
system_monitor/
├── cpu_20230615_143022.csv
├── memory_20230615_143022.csv
├── disk_20230615_143022.csv
├── application_20230615_143022.csv
└── monitor.log
2.3.2. 完整实践代码
import psutil
import platform
import time
from datetime import datetime
import csv
import os
from pathlib import Path# 配置监控参数
MONITOR_DIR = "system_monitor" # 监控文件存储目录
LOG_FILE = "monitor.log" # 日志文件
MAX_FILE_SIZE = 10 * 1024 * 1024 # 单个CSV文件最大10MB
FLUSH_INTERVAL = 60 # 每60秒刷新一次文件class MonitorWriter:def __init__(self):self.base_dir = Path(MONITOR_DIR)self.base_dir.mkdir(exist_ok=True)self.writers = {}self.files = {}self.last_flush = time.time()# 初始化日志文件self.log_file = self.base_dir / LOG_FILEself.setup_logging()def setup_logging(self):"""设置日志文件"""with open(self.log_file, 'a') as f:f.write(f"\n=== 监控开始于 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ===\n")def get_writer(self, category):"""获取指定类别的CSV写入器"""if category not in self.writers:# 创建新文件（带时间戳）timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")filename = self.base_dir / f"{category}_{timestamp}.csv"# 写入CSV头file = open(filename, 'a', newline='')writer = csv.DictWriter(file, fieldnames=self.get_headers(category))if file.tell() == 0:writer.writeheader()self.files[category] = fileself.writers[category] = writerreturn self.writers[category]def get_headers(self, category):"""获取各类别的CSV表头"""headers = ["timestamp"]if category == "cpu":headers.extend(["total_usage", *[f"core_{i}" for i in range(psutil.cpu_count())]])elif category == "memory":headers.extend(["total", "available", "used", "free", "percent", "swap_total", "swap_used", "swap_percent"])elif category == "disk":headers.extend(["device", "mountpoint", "total", "used", "free", "percent"])elif category == "application":headers.extend(["pid", "name", "cpu_percent", "memory_rss", "memory_percent", "threads"])return headersdef write_data(self, category, data):"""写入监控数据"""try:writer = self.get_writer(category)writer.writerow(data)# 定期刷新文件和检查大小if time.time() - self.last_flush