Python实现系统性能监控：基础参数获取与文件存储方案

引言

在系统运维和开发过程中，实时监控系统性能参数是保障服务稳定性的关键环节。无论是服务器资源管理、应用程序优化，还是故障排查，都需要准确获取CPU使用率、内存占用、磁盘I/O等基础指标。Python凭借其丰富的标准库和第三方模块，能够高效实现这一需求。本文将详细介绍如何使用Python获取系统基础性能参数，并将这些数据结构化存储到文件中，为后续分析提供可靠数据源。

一、系统性能参数获取的核心方法

1.1 CPU使用率监控

CPU是系统运行的核心资源，其使用率直接反映系统负载情况。Python可通过psutil库获取CPU信息：

import psutil
def get_cpu_info():
    cpu_percent = psutil.cpu_percent(interval=1)  # 获取1秒内的CPU使用率
    cpu_count = psutil.cpu_count()  # 逻辑CPU核心数
    cpu_freq = psutil.cpu_freq()  # CPU频率（MHz）
    return {
        'usage_percent': cpu_percent,
        'core_count': cpu_count,
        'current_freq': cpu_freq.current if cpu_freq else None
    }

关键点：

• interval参数控制采样周期，避免瞬时值波动
• 多核系统需注意区分逻辑核心与物理核心
• 频率信息可辅助判断CPU是否处于降频状态

1.2 内存使用分析

内存监控包括物理内存和交换分区使用情况：

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

数据解读：

• available比free更能反映实际可用内存（包含缓存回收）
• 交换分区使用率过高可能引发性能问题
• 建议设置阈值告警（如持续>80%）

1.3 磁盘I/O监控

磁盘性能直接影响数据读写效率：

def get_disk_info():
    partitions = psutil.disk_partitions()
    usage_stats = []
    for part in partitions:
        try:
            usage = psutil.disk_usage(part.mountpoint)
            io = psutil.disk_io_counters(perdisk=True).get(part.device.split('/')[-1], None)
            usage_stats.append({
                'device': part.device,
                'mountpoint': part.mountpoint,
                'total': usage.total,
                'used_percent': usage.percent,
                'read_bytes': io.read_bytes if io else 0,
                'write_bytes': io.write_bytes if io else 0
            })
        except PermissionError:
            continue
    return usage_stats

实施建议：

• 监控根分区和关键数据分区
• 结合read_bytes/write_bytes分析I/O压力
• 定期检查inode使用情况（通过os.statvfs）

1.4 网络流量统计

网络性能监控包含带宽使用和网络错误统计：

def get_network_info():
    net_io = psutil.net_io_counters()
    connections = psutil.net_connections(kind='inet')
    return {
        'bytes_sent': net_io.bytes_sent,
        'bytes_recv': net_io.bytes_recv,
        'packets_sent': net_io.packets_sent,
        'packets_recv': net_io.packets_recv,
        'active_connections': len(connections)
    }

优化方向：

• 区分内外网流量（通过连接地址过滤）
• 计算瞬时速率（Δbytes/Δtime）
• 监控TCP重传和错误计数

二、数据持久化存储方案

2.1 CSV格式存储

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

import csv
from datetime import datetime
def write_to_csv(data, filename='system_metrics.csv'):
    fieldnames = ['timestamp'] + list(data[0].keys()) if data else ['timestamp']
    with open(filename, 'a', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if f.tell() == 0:  # 文件为空时写入表头
            writer.writeheader()
        for entry in data:
            writer.writerow({**{'timestamp': datetime.now().isoformat()}, **entry})

优势：

• 通用性强，所有系统均可读取
• 支持追加写入模式
• 适合长期历史数据存储

2.2 JSON格式存储

适合程序间数据交换和复杂结构：

import json
def write_to_json(data, filename='system_metrics.json'):
    with open(filename, 'a') as f:
        entry = {
            'timestamp': datetime.now().isoformat(),
            'metrics': data
        }
        f.write(json.dumps(entry) + '\n')  # 每行一个JSON对象

适用场景：

• 需要保留数据嵌套结构时
• 与Web服务或NoSQL数据库集成
• 需要人类可读的调试信息时

2.3 SQLite数据库存储

适合需要查询和索引的场景：

import sqlite3
def init_db(db_file='system_metrics.db'):
    conn = sqlite3.connect(db_file)
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS metrics
                 (timestamp TEXT PRIMARY KEY,
                  cpu_usage REAL,
                  mem_used_percent REAL,
                  disk_used_percent REAL,
                  net_bytes_sent INTEGER,
                  net_bytes_recv INTEGER)''')
    conn.commit()
    conn.close()
def insert_metrics(db_file, metrics):
    conn = sqlite3.connect(db_file)
    c = conn.cursor()
    c.execute('''INSERT INTO metrics VALUES
                 (?, ?, ?, ?, ?, ?)''',
              (datetime.now().isoformat(),
               metrics['usage_percent'],
               metrics['used_percent'],
               metrics['used_percent'] if 'used_percent' in metrics else None,  # 示例适配
               metrics['bytes_sent'] if 'bytes_sent' in metrics else 0,
               metrics['bytes_recv'] if 'bytes_recv' in metrics else 0))
    conn.commit()
    conn.close()

性能优化：

• 批量插入时使用事务
• 为常用查询字段创建索引
• 定期执行VACUUM命令整理碎片

三、完整实现示例

3.1 综合监控脚本

import time
import psutil
from datetime import datetime
def collect_metrics():
    # CPU信息
    cpu_info = {
        'cpu_usage': psutil.cpu_percent(interval=1),
        'cpu_count': psutil.cpu_count()
    }
    # 内存信息
    mem = psutil.virtual_memory()
    mem_info = {
        'mem_total': mem.total,
        'mem_used_percent': mem.percent
    }
    # 磁盘信息（仅监控根分区）
    try:
        disk = psutil.disk_usage('/')
        disk_info = {
            'disk_total': disk.total,
            'disk_used_percent': disk.percent
        }
    except PermissionError:
        disk_info = {}
    # 网络信息
    net_io = psutil.net_io_counters()
    net_info = {
        'net_bytes_sent': net_io.bytes_sent,
        'net_bytes_recv': net_io.bytes_recv
    }
    return {**cpu_info, **mem_info, **disk_info, **net_info}
def write_metrics_to_csv(metrics, filename='metrics.csv'):
    import csv
    fieldnames = ['timestamp'] + [k for k in metrics.keys() if k != 'timestamp']
    with open(filename, 'a', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if f.tell() == 0:
            writer.writeheader()
        writer.writerow({**{'timestamp': datetime.now().isoformat()}, **metrics})
def main(interval=60, duration=3600):
    end_time = time.time() + duration
    while time.time() < end_time:
        metrics = collect_metrics()
        write_metrics_to_csv(metrics)
        print(f"Collected metrics at {metrics['timestamp']}")
        time.sleep(interval)
if __name__ == "__main__":
    main(interval=10, duration=600)  # 每10秒采集一次，持续10分钟

3.2 高级功能扩展

1. 异常检测：在采集时加入阈值检查

   def check_thresholds(metrics):
    alerts = []
    if metrics['cpu_usage'] > 90:
        alerts.append("CPU过载")
    if metrics['mem_used_percent'] > 85:
        alerts.append("内存不足")
    return alerts

2. 数据压缩：长期存储时使用zlib压缩
   ```python
   import zlib

def compress_data(data):
return zlib.compress(str(data).encode(‘utf-8’))

3. **远程传输**：集成HTTP POST发送到监控服务器
```python
import requests
def send_to_server(data, url):
    try:
        requests.post(url, json=data, timeout=5)
    except requests.exceptions.RequestException as e:
        print(f"发送失败: {e}")

四、最佳实践建议

1. 采样频率选择：

   • 短期调试：1-5秒
   • 长期监控：30-300秒
   • 避免过高频率导致性能自干扰

2. 数据存储策略：

   • 原始数据保留30天
   • 聚合数据（如每小时平均值）长期保存
   • 定期归档旧数据到冷存储

3. 安全考虑：

   • 敏感数据（如内存内容）避免记录
   • 数据库文件设置适当权限
   • 网络传输使用HTTPS

4. 性能优化：

   • 异步采集避免阻塞主程序
   • 批量写入减少I/O操作
   • 内存缓存临时数据

五、常见问题解决方案

1. 权限不足错误：

   • 使用管理员权限运行脚本
   • 或配置psutil的权限参数

2. 数据丢失问题：

   • 实现写入失败的重试机制
   • 添加校验和验证数据完整性

3. 跨平台兼容性：

   • 检测操作系统类型
   • 针对Windows/Linux使用不同的分区检测逻辑

4. 时间同步问题：

   • 使用NTP服务保持系统时间准确
   • 记录时区信息

结论

通过Python实现系统性能参数的自动化采集和结构化存储，可以构建轻量级但功能完善的监控系统。本文介绍的方案结合了psutil库的强大功能与多种存储格式的灵活选择，能够满足从个人开发到企业级监控的不同需求。实际部署时，建议根据具体场景调整采样频率、存储周期和告警阈值，并考虑将数据可视化展示以提升监控效率。随着系统规模的扩大，可进一步集成Prometheus、Grafana等专业工具构建更完善的监控体系。