Python精准监控：获取系统性能参数并写入文件全攻略

一、技术背景与核心价值

在云计算、大数据和容器化部署普及的今天，系统性能监控已成为运维工作的核心环节。Python凭借其跨平台特性、丰富的第三方库和简洁的语法，成为实现系统监控的理想工具。通过获取CPU使用率、内存占用、磁盘I/O等关键指标，开发者可以：

1. 实时诊断系统瓶颈
2. 预测硬件资源需求
3. 构建自动化运维系统
4. 生成可视化性能报告

相较于传统监控工具（如Zabbix、Nagios），Python方案具有轻量级、可定制化的优势，特别适合中小规模系统或嵌入式环境。

二、核心性能参数解析与获取方法

1. CPU性能指标采集

CPU监控包含三个关键维度：

• 整体使用率：反映CPU繁忙程度
• 核心使用率：识别多核系统中的负载不均衡
• 上下文切换次数：诊断进程调度异常

实现方案：

import psutil
def get_cpu_metrics():
    # 获取CPU整体使用率（间隔1秒采样）
    cpu_percent = psutil.cpu_percent(interval=1)
    # 获取各核心使用率
    cpu_cores = psutil.cpu_percent(interval=1, percpu=True)
    # 获取上下文切换次数（需Linux系统）
    try:
        with open('/proc/stat') as f:
            for line in f:
                if line.startswith('ctxt'):
                    ctxt_switches = int(line.split()[1])
                    break
    except:
        ctxt_switches = None
    return {
        'total_usage': cpu_percent,
        'core_usages': cpu_cores,
        'context_switches': ctxt_switches
    }

2. 内存性能监控

内存监控需区分物理内存和交换分区：

• 可用内存：available字段更准确反映实际可用内存
• 缓存占用：cached和buffers的合理利用
• 交换活动：频繁交换可能预示内存不足

实现方案：

def get_memory_metrics():
    mem = psutil.virtual_memory()
    swap = psutil.swap_memory()
    return {
        'total': mem.total,
        'available': mem.available,
        'used_percent': mem.percent,
        'swap_used': swap.used,
        'swap_percent': swap.percent
    }

3. 磁盘I/O监控

关键监控点包括：

• 读写速度：识别I/O密集型进程
• 磁盘利用率：检测磁盘饱和度
• 等待队列：诊断I/O瓶颈

实现方案：

def get_disk_metrics():
    # 获取磁盘使用情况
    disk_usage = psutil.disk_usage('/')
    # 获取磁盘I/O统计（需Linux系统）
    try:
        with open('/proc/diskstats') as f:
            # 解析磁盘I/O数据（示例为sda设备）
            for line in f:
                if 'sda' in line:
                    parts = line.split()
                    reads = int(parts[3])  # 读完成次数
                    writes = int(parts[7])  # 写完成次数
                    read_sectors = int(parts[5])  # 读扇区数
                    write_sectors = int(parts[9])  # 写扇区数
                    break
    except:
        reads = writes = read_sectors = write_sectors = None
    return {
        'usage_percent': disk_usage.percent,
        'reads': reads,
        'writes': writes,
        'read_sectors': read_sectors,
        'write_sectors': write_sectors
    }

三、数据持久化方案

1. CSV文件存储

适合结构化数据存储和Excel分析：

import csv
from datetime import datetime
def write_to_csv(data, filename='system_metrics.csv'):
    timestamp = datetime.now().isoformat()
    row = {
        'timestamp': timestamp,
        'cpu_total': data['cpu']['total_usage'],
        'mem_used_percent': data['memory']['used_percent'],
        'disk_usage_percent': data['disk']['usage_percent']
        # 可扩展更多字段
    }
    # 写入文件（追加模式）
    with open(filename, 'a', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=row.keys())
        if f.tell() == 0:  # 文件为空时写入表头
            writer.writeheader()
        writer.writerow(row)

2. JSON文件存储

适合结构化数据和API交互：

import json
def write_to_json(data, filename='system_metrics.json'):
    timestamp = datetime.now().isoformat()
    metric_data = {
        'timestamp': timestamp,
        'cpu': data['cpu'],
        'memory': data['memory'],
        'disk': data['disk']
    }
    # 读取现有数据（如果存在）
    try:
        with open(filename, 'r') as f:
            existing_data = json.load(f)
    except FileNotFoundError:
        existing_data = []
    # 追加新数据
    existing_data.append(metric_data)
    # 写入文件
    with open(filename, 'w') as f:
        json.dump(existing_data, f, indent=2)

3. 数据库存储方案

对于长期监控，建议使用SQLite或InfluxDB：

import sqlite3
def init_db():
    conn = sqlite3.connect('system_metrics.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS metrics
                 (timestamp TEXT PRIMARY KEY,
                  cpu_total REAL,
                  mem_used_percent REAL,
                  disk_usage_percent REAL)''')
    conn.commit()
    conn.close()
def write_to_db(data):
    conn = sqlite3.connect('system_metrics.db')
    c = conn.cursor()
    timestamp = datetime.now().isoformat()
    c.execute('''INSERT INTO metrics VALUES
                 (?, ?, ?, ?)''',
              (timestamp,
               data['cpu']['total_usage'],
               data['memory']['used_percent'],
               data['disk']['usage_percent']))
    conn.commit()
    conn.close()

四、完整实现示例

import time
import psutil
from datetime import datetime
import json
def collect_metrics():
    return {
        'cpu': get_cpu_metrics(),
        'memory': get_memory_metrics(),
        'disk': get_disk_metrics()
    }
def save_metrics(data, format='json'):
    timestamp = datetime.now().isoformat()
    metric_entry = {
        'timestamp': timestamp,
        'metrics': data
    }
    if format == 'json':
        try:
            with open('metrics.json', 'r') as f:
                existing = json.load(f)
        except FileNotFoundError:
            existing = []
        existing.append(metric_entry)
        with open('metrics.json', 'w') as f:
            json.dump(existing, f, indent=2)
    elif format == 'csv':
        # CSV实现略（参考前文）
        pass
def main(interval=5, duration=60):
    end_time = time.time() + duration
    while time.time() < end_time:
        data = collect_metrics()
        save_metrics(data)
        print(f"Collected metrics at {datetime.now().isoformat()}")
        time.sleep(interval)
if __name__ == '__main__':
    main(interval=5, duration=60)  # 每5秒采集一次，持续60秒

五、优化建议与扩展方向

1. 性能优化：

   • 使用异步IO减少采集延迟
   • 对多核系统采用并行采集
   • 实现增量存储避免重复数据

2. 功能扩展：

   • 添加网络带宽监控
   • 实现异常阈值告警
   • 开发Web可视化界面

3. 跨平台适配：

   • Windows系统使用wmi模块替代/proc文件系统
   • macOS系统使用iostat命令解析

4. 安全考虑：

   • 对输出文件进行权限控制
   • 敏感数据加密存储
   • 实现采集频率限制

六、典型应用场景

1. 服务器健康检查：定期生成性能报告
2. 性能基准测试：对比不同配置下的系统表现
3. 故障排查：事故发生时回溯性能数据
4. 容量规划：基于历史数据预测资源需求

通过Python实现系统性能监控，开发者可以获得灵活、高效且可定制的监控解决方案。结合适当的存储策略，这些数据不仅可以用于实时监控，更能为系统优化和故障预防提供有力支持。在实际部署时，建议根据具体需求调整采集频率和数据粒度，在监控精度和系统负载之间取得平衡。