logo

开发者热搜

Python实时监控系统性能：参数获取与文件存储全流程解析

作者：沙与沫2025.09.25 23:05浏览量：5

简介：本文详细讲解如何使用Python获取系统基础性能参数（CPU、内存、磁盘、网络等），并将数据结构化写入文件。通过标准库和第三方工具实现跨平台监控，提供完整代码示例与异常处理方案。

Python获取系统基础性能参数实现写入文件

在系统运维和性能分析场景中，实时获取并记录系统基础性能参数是故障排查、容量规划和性能优化的关键环节。Python凭借其丰富的标准库和第三方生态，能够高效实现跨平台的系统监控与数据持久化。本文将系统阐述如何通过Python获取CPU、内存、磁盘、网络等核心性能指标，并将数据结构化写入文件，提供完整的实现方案与优化建议。

一、核心性能参数分类与获取方式

1. CPU使用率监控

CPU作为计算核心，其使用率直接反映系统负载。Python可通过以下方式获取：

  • psutil库：跨平台系统监控工具，支持获取CPU逻辑核心数、使用率、空闲时间等
  • /proc/stat文件（Linux）：解析内核提供的CPU时间统计
  • wmi模块（Windows）：通过WMI接口获取性能计数器
  1. import psutil
  3. def get_cpu_info():
  4.     # 获取CPU逻辑核心数
  5.     cpu_count = psutil.cpu_count(logical=True)
  6.     # 获取每颗CPU的使用率（间隔1秒采样）
  7.     cpu_percent = psutil.cpu_percent(interval=1, percpu=True)
  8.     # 获取系统全局CPU使用率
  9.     global_percent = psutil.cpu_percent(interval=1)
  11.     return {
  12.         "cpu_cores": cpu_count,
  13.         "percpu_usage": cpu_percent,
  14.         "global_usage": global_percent
  15.     }

2. 内存使用情况

内存监控需关注总量、已用、空闲和缓存等指标：

  • psutil.virtual_memory()：获取物理内存信息
  • psutil.swap_memory()：获取交换分区信息
  • memory_profiler：更精细的内存分析（需安装）
  1. def get_memory_info():
  2.     mem = psutil.virtual_memory()
  3.     swap = psutil.swap_memory()
  5.     return {
  6.         "total_memory": mem.total,
  7.         "available": mem.available,
  8.         "used_percent": mem.percent,
  9.         "swap_total": swap.total,
  10.         "swap_used_percent": swap.percent
  11.     }

3. 磁盘I/O监控

磁盘性能直接影响系统响应速度，需监控：

  • 分区使用情况：psutil.disk_usage()
  • 磁盘I/O统计：psutil.disk_io_counters()
  • 文件系统信息：os.statvfs()（Linux）
  1. def get_disk_info():
  2.     # 获取所有磁盘分区
  3.     partitions = psutil.disk_partitions(all=False)
  4.     disk_info = []
  6.     for partition in partitions:
  7.         usage = psutil.disk_usage(partition.mountpoint)
  8.         io_stats = psutil.disk_io_counters(perdisk=True).get(partition.device.split('/')[-1], None)
  10.         disk_info.append({
  11.             "device": partition.device,
  12.             "mountpoint": partition.mountpoint,
  13.             "fstype": partition.fstype,
  14.             "total": usage.total,
  15.             "used": usage.used,
  16.             "free": usage.free,
  17.             "percent": usage.percent,
  18.             # 转换为MB单位
  19.             "read_bytes": io_stats.read_bytes / (1024**2) if io_stats else 0,
  20.             "write_bytes": io_stats.write_bytes / (1024**2) if io_stats else 0
  21.         })
  23.     return disk_info

4. 网络流量统计

网络监控需捕获：

  • 接口流量：psutil.net_io_counters()
  • 连接状态：psutil.net_connections()
  • 带宽使用率：需结合时间差计算
  1. def get_network_info():
  2.     net_io = psutil.net_io_counters()
  3.     connections = psutil.net_connections(kind='inet')
  5.     return {
  6.         "bytes_sent": net_io.bytes_sent / (1024**2),  # 转换为MB
  7.         "bytes_recv": net_io.bytes_recv / (1024**2),
  8.         "packets_sent": net_io.packets_sent,
  9.         "packets_recv": net_io.packets_recv,
  10.         "active_connections": len(connections)
  11.     }

二、数据持久化方案

1. CSV文件存储

适合结构化数据存储，便于后续分析：

  1. import csv
  2. from datetime import datetime
  4. def write_to_csv(data, filename="system_metrics.csv"):
  5.     # 如果是首次写入，创建文件并写入表头
  6.     first_write = not os.path.exists(filename)
  8.     with open(filename, mode='a', newline='') as file:
  9.         writer = csv.writer(file)
  11.         if first_write:
  12.             writer.writerow(["timestamp", "cpu_global", "mem_used%", "disk_used%", "net_sent(MB)", "net_recv(MB)"])
  14.         # 提取关键指标写入
  15.         writer.writerow([
  16.             datetime.now().isoformat(),
  17.             data["cpu"]["global_usage"],
  18.             data["memory"]["used_percent"],
  19.             sum(disk["percent"] for disk in data["disk"])/len(data["disk"]) if data["disk"] else 0,
  20.             data["network"]["bytes_sent"],
  21.             data["network"]["bytes_recv"]
  22.         ])

2. JSON文件存储

适合嵌套数据结构，保留完整信息：

  1. import json
  3. def write_to_json(data, filename="system_metrics.json"):
  4.     # 读取现有数据（如果存在）
  5.     existing_data = []
  6.     if os.path.exists(filename):
  7.         with open(filename, 'r') as f:
  8.             try:
  9.                 existing_data = json.load(f)
  10.             except json.JSONDecodeError:
  11.                 existing_data = []
  13.     # 添加新记录
  14.     new_entry = {
  15.         "timestamp": datetime.now().isoformat(),
  16.         "cpu": data["cpu"],
  17.         "memory": data["memory"],
  18.         "disk": data["disk"],
  19.         "network": data["network"]
  20.     }
  21.     existing_data.append(new_entry)
  23.     # 写入文件
  24.     with open(filename, 'w') as f:
  25.         json.dump(existing_data, f, indent=2)

3. 数据库存储（SQLite示例）

适合长期存储和复杂查询：

  1. import sqlite3
  3. def init_db(db_path="system_metrics.db"):
  4.     conn = sqlite3.connect(db_path)
  5.     cursor = conn.cursor()
  7.     cursor.execute('''
  8.         CREATE TABLE IF NOT EXISTS metrics (
  9.             id INTEGER PRIMARY KEY AUTOINCREMENT,
  10.             timestamp TEXT NOT NULL,
  11.             cpu_global REAL,
  12.             mem_used_percent REAL,
  13.             disk_avg_used_percent REAL,
  14.             net_sent_mb REAL,
  15.             net_recv_mb REAL
  16.         )
  17.     ''')
  19.     conn.commit()
  20.     conn.close()
  22. def write_to_db(data, db_path="system_metrics.db"):
  23.     conn = sqlite3.connect(db_path)
  24.     cursor = conn.cursor()
  26.     cursor.execute('''
  27.         INSERT INTO metrics 
  28.         (timestamp, cpu_global, mem_used_percent, disk_avg_used_percent, net_sent_mb, net_recv_mb)
  29.         VALUES (?, ?, ?, ?, ?, ?)
  30.     ''', (
  31.         datetime.now().isoformat(),
  32.         data["cpu"]["global_usage"],
  33.         data["memory"]["used_percent"],
  34.         sum(disk["percent"] for disk in data["disk"])/len(data["disk"]) if data["disk"] else 0,
  35.         data["network"]["bytes_sent"],
  36.         data["network"]["bytes_recv"]
  37.     ))
  39.     conn.commit()
  40.     conn.close()

三、完整实现示例

  1. import os
  2. import time
  3. import psutil
  4. from datetime import datetime
  5. import json
  6. import csv
  7. import sqlite3
  9. def collect_system_metrics():
  10.     # CPU信息
  11.     cpu_info = {
  12.         "cpu_cores": psutil.cpu_count(logical=True),
  13.         "percpu_usage": psutil.cpu_percent(interval=1, percpu=True),
  14.         "global_usage": psutil.cpu_percent(interval=1)
  15.     }
  17.     # 内存信息
  18.     mem = psutil.virtual_memory()
  19.     swap = psutil.swap_memory()
  20.     memory_info = {
  21.         "total_memory": mem.total,
  22.         "available": mem.available,
  23.         "used_percent": mem.percent,
  24.         "swap_total": swap.total,
  25.         "swap_used_percent": swap.percent
  26.     }
  28.     # 磁盘信息
  29.     partitions = psutil.disk_partitions(all=False)
  30.     disk_info = []
  31.     for partition in partitions:
  32.         try:
  33.             usage = psutil.disk_usage(partition.mountpoint)
  34.             io_stats = psutil.disk_io_counters(perdisk=True).get(partition.device.split('/')[-1], None)
  35.             disk_info.append({
  36.                 "device": partition.device,
  37.                 "mountpoint": partition.mountpoint,
  38.                 "fstype": partition.fstype,
  39.                 "total": usage.total,
  40.                 "used": usage.used,
  41.                 "free": usage.free,
  42.                 "percent": usage.percent,
  43.                 "read_bytes": io_stats.read_bytes / (1024**2) if io_stats else 0,
  44.                 "write_bytes": io_stats.write_bytes / (1024**2) if io_stats else 0
  45.             })
  46.         except PermissionError:
  47.             continue
  49.     # 网络信息
  50.     net_io = psutil.net_io_counters()
  51.     connections = psutil.net_connections(kind='inet')
  52.     network_info = {
  53.         "bytes_sent": net_io.bytes_sent / (1024**2),
  54.         "bytes_recv": net_io.bytes_recv / (1024**2),
  55.         "packets_sent": net_io.packets_sent,
  56.         "packets_recv": net_io.packets_recv,
  57.         "active_connections": len(connections)
  58.     }
  60.     return {
  61.         "timestamp": datetime.now().isoformat(),
  62.         "cpu": cpu_info,
  63.         "memory": memory_info,
  64.         "disk": disk_info,
  65.         "network": network_info
  66.     }
  68. def save_metrics(data, csv_file="metrics.csv", json_file="metrics.json", db_file="metrics.db"):
  69.     # 初始化数据库
  70.     if not os.path.exists(db_file):
  71.         conn = sqlite3.connect(db_file)
  72.         cursor = conn.cursor()
  73.         cursor.execute('''
  74.             CREATE TABLE metrics (
  75.                 id INTEGER PRIMARY KEY AUTOINCREMENT,
  76.                 timestamp TEXT NOT NULL,
  77.                 cpu_global REAL,
  78.                 mem_used_percent REAL,
  79.                 disk_avg_used_percent REAL,
  80.                 net_sent_mb REAL,
  81.                 net_recv_mb REAL
  82.             )
  83.         ''')
  84.         conn.commit()
  85.         conn.close()
  87.     # 写入CSV
  88.     first_write = not os.path.exists(csv_file)
  89.     with open(csv_file, mode='a', newline='') as file:
  90.         writer = csv.writer(file)
  91.         if first_write:
  92.             writer.writerow(["timestamp", "cpu_global", "mem_used%", "disk_used%", "net_sent(MB)", "net_recv(MB)"])
  94.         writer.writerow([
  95.             data["timestamp"],
  96.             data["cpu"]["global_usage"],
  97.             data["memory"]["used_percent"],
  98.             sum(disk["percent"] for disk in data["disk"])/len(data["disk"]) if data["disk"] else 0,
  99.             data["network"]["bytes_sent"],
  100.             data["network"]["bytes_recv"]
  101.         ])
  103.     # 写入JSON
  104.     existing_data = []
  105.     if os.path.exists(json_file):
  106.         with open(json_file, 'r') as f:
  107.             try:
  108.                 existing_data = json.load(f)
  109.             except json.JSONDecodeError:
  110.                 existing_data = []
  111.     existing_data.append(data)
  112.     with open(json_file, 'w') as f:
  113.         json.dump(existing_data, f, indent=2)
  115.     # 写入数据库
  116.     conn = sqlite3.connect(db_file)
  117.     cursor = conn.cursor()
  118.     cursor.execute('''
  119.         INSERT INTO metrics 
  120.         (timestamp, cpu_global, mem_used_percent, disk_avg_used_percent, net_sent_mb, net_recv_mb)
  121.         VALUES (?, ?, ?, ?, ?, ?)
  122.     ''', (
  123.         data["timestamp"],
  124.         data["cpu"]["global_usage"],
  125.         data["memory"]["used_percent"],
  126.         sum(disk["percent"] for disk in data["disk"])/len(data["disk"]) if data["disk"] else 0,
  127.         data["network"]["bytes_sent"],
  128.         data["network"]["bytes_recv"]
  129.     ))
  130.     conn.commit()
  131.     conn.close()
  133. if __name__ == "__main__":
  134.     # 首次运行初始化数据库
  135.     init_db()
  137.     try:
  138.         while True:
  139.             metrics = collect_system_metrics()
  140.             save_metrics(metrics)
  141.             print(f"Metrics collected at {metrics['timestamp']}")
  142.             time.sleep(60)  # 每分钟采集一次
  143.     except KeyboardInterrupt:
  144.         print("Monitoring stopped by user")

四、优化建议与注意事项

  1. 性能影响：高频采集（如<1秒间隔）可能影响系统性能，建议根据需求调整采样频率
  2. 权限问题：Linux下获取某些磁盘信息可能需要root权限
  3. 跨平台兼容：使用psutil可最大限度保证代码跨平台性，但某些高级功能可能平台受限
  4. 数据旋转：长期运行需实现日志轮转，避免单个文件过大
  5. 异常处理：添加完善的异常捕获，特别是文件操作和网络接口
  6. 可视化建议：可将数据导入Grafana或Python的Matplotlib/Plotly进行可视化分析

五、扩展应用场景

  1. 自动化运维：集成到Ansible/SaltStack等自动化工具中
  2. 容器监控：在Docker/Kubernetes环境中监控容器资源使用
  3. 性能基准测试：记录系统在不同负载下的性能表现
  4. 异常检测：通过分析历史数据建立基线，检测异常资源使用

通过本文的实现方案，开发者可以快速构建一个轻量级的系统监控工具，既能满足即时性能分析需求，也可作为更复杂监控系统的基础组件。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数
活动
咨询