logo

开发者热搜

Python精准监控：系统性能参数获取与文件存储实战指南

作者：有好多问题2025.09.25 23:05浏览量：0

简介：本文详细介绍如何使用Python获取系统CPU、内存、磁盘等基础性能参数，并通过文件操作实现数据持久化存储，为系统监控和性能分析提供实用方案。

Python精准监控：系统性能参数获取与文件存储实战指南

在系统运维和开发过程中，实时获取并分析系统性能参数是保障服务稳定性的关键环节。Python凭借其丰富的标准库和第三方生态，能够高效实现系统性能监控数据的采集与存储。本文将系统阐述如何使用Python获取CPU使用率、内存占用、磁盘I/O等基础性能指标，并将数据结构化写入文件，为后续的性能分析和自动化运维提供数据支撑。

一、系统性能参数采集的核心方法

1. CPU使用率采集

CPU使用率是衡量系统负载的核心指标，Python可通过psutil库实现跨平台采集。psutil提供了cpu_percent()方法，支持按核心或全局统计：

  1. import psutil
  3. def get_cpu_usage():
  4.     # 获取所有CPU核心的使用率（列表形式）
  5.     cpu_percent_per_core = psutil.cpu_percent(interval=1, percpu=True)
  6.     # 获取全局CPU使用率
  7.     cpu_percent_total = psutil.cpu_percent(interval=1)
  8.     return {
  9.         "total_usage": cpu_percent_total,
  10.         "per_core_usage": cpu_percent_per_core
  11.     }

关键参数说明：

  • interval=1：采样间隔（秒），确保数据准确性
  • percpu=True：返回各核心使用率，否则返回全局值

2. 内存信息采集

内存状态直接影响系统运行效率，psutil.virtual_memory()可获取详细内存信息：

  1. def get_memory_info():
  2.     mem = psutil.virtual_memory()
  3.     return {
  4.         "total": mem.total,          # 总内存（字节）
  5.         "available": mem.available,  # 可用内存（字节）
  6.         "used": mem.used,            # 已用内存（字节）
  7.         "percent": mem.percent       # 内存使用率（%）
  8.     }

数据解析要点：

  • available字段比free更准确，包含缓存和缓冲区可回收内存
  • 字节单位需转换为GB/MB便于阅读（1GB=1024^3字节）

3. 磁盘I/O与空间监控

磁盘性能是系统瓶颈的常见来源，需同时监控使用情况和I/O负载：

  1. def get_disk_info():
  2.     # 磁盘分区使用情况
  3.     disk_usage = psutil.disk_usage('/')
  4.     # 磁盘I/O统计（需结合采样间隔计算速率）
  5.     io_before = psutil.disk_io_counters()
  6.     time.sleep(1)  # 间隔1秒
  7.     io_after = psutil.disk_io_counters()
  9.     read_speed = (io_after.read_bytes - io_before.read_bytes) / 1024  # KB/s
  10.     write_speed = (io_after.write_bytes - io_before.write_bytes) / 1024
  12.     return {
  13.         "usage": {
  14.             "total": disk_usage.total,
  15.             "used": disk_usage.used,
  16.             "free": disk_usage.free,
  17.             "percent": disk_usage.percent
  18.         },
  19.         "io_speed": {
  20.             "read_kbps": read_speed,
  21.             "write_kbps": write_speed
  22.         }
  23.     }

实现注意事项：

  • 磁盘I/O速率需通过两次采样差值计算
  • 路径'/'可根据需要替换为其他分区

二、性能数据结构化存储方案

1. CSV文件存储实现

CSV格式适合结构化数据存储，可使用Python内置csv模块：

  1. import csv
  2. from datetime import datetime
  4. def write_to_csv(data, filename="system_metrics.csv"):
  5.     timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
  6.     flat_data = {
  7.         "timestamp": timestamp,
  8.         "cpu_total": data["cpu"]["total_usage"],
  9.         "cpu_core1": data["cpu"]["per_core_usage"][0] if len(data["cpu"]["per_core_usage"]) > 0 else 0,
  10.         "mem_used_percent": data["memory"]["percent"],
  11.         "disk_used_percent": data["disk"]["usage"]["percent"],
  12.         "disk_read_kbps": data["disk"]["io_speed"]["read_kbps"],
  13.         "disk_write_kbps": data["disk"]["io_speed"]["write_kbps"]
  14.     }
  16.     fieldnames = ["timestamp", "cpu_total", "cpu_core1", "mem_used_percent", 
  17.                   "disk_used_percent", "disk_read_kbps", "disk_write_kbps"]
  19.     # 写入模式（追加）
  20.     with open(filename, mode='a', newline='') as file:
  21.         writer = csv.DictWriter(file, fieldnames=fieldnames)
  22.         # 仅在文件为空时写入表头
  23.         if file.tell() == 0:
  24.             writer.writeheader()
  25.         writer.writerow(flat_data)

优化建议：

  • 使用mode='a'实现数据追加
  • 通过file.tell()判断是否为空文件，避免重复写入表头
  • 添加时间戳字段实现时序数据分析

2. JSON文件存储方案

JSON格式适合嵌套数据存储，可使用json模块：

  1. import json
  3. def write_to_json(data, filename="system_metrics.json"):
  4.     timestamp = datetime.now().isoformat()
  5.     full_data = {
  6.         "timestamp": timestamp,
  7.         "metrics": data
  8.     }
  10.     # 读取现有数据（如果存在）
  11.     try:
  12.         with open(filename, 'r') as file:
  13.             existing_data = json.load(file)
  14.     except FileNotFoundError:
  15.         existing_data = []
  17.     # 追加新数据
  18.     existing_data.append(full_data)
  20.     # 写入文件
  21.     with open(filename, 'w') as file:
  22.         json.dump(existing_data, file, indent=4)

适用场景：

  • 需要保留完整数据结构的场景
  • 后续需用Python或其他语言解析的场景
  • 数据量较小（JSON不适合超大文件）

三、完整监控系统实现

1. 主监控程序框架

  1. import time
  2. import psutil
  3. from datetime import datetime
  5. def collect_metrics():
  6.     return {
  7.         "cpu": get_cpu_usage(),
  8.         "memory": get_memory_info(),
  9.         "disk": get_disk_info()
  10.     }
  12. def main(interval=5, output_format="csv"):
  13.     while True:
  14.         metrics = collect_metrics()
  15.         if output_format == "csv":
  16.             write_to_csv(metrics)
  17.         elif output_format == "json":
  18.             write_to_json(metrics)
  19.         print(f"[{datetime.now()}] Metrics collected and saved")
  20.         time.sleep(interval)
  22. if __name__ == "__main__":
  23.     main(interval=5, output_format="csv")

2. 高级功能扩展

（1）异常处理机制

  1. def safe_collect_metrics():
  2.     try:
  3.         return collect_metrics()
  4.     except Exception as e:
  5.         error_data = {
  6.             "timestamp": datetime.now().isoformat(),
  7.             "error": str(e),
  8.             "metrics": None
  9.         }
  10.         write_to_json(error_data, "error_log.json")
  11.         return None

（2）数据压缩存储

对于长期监控，可使用gzip压缩JSON文件：

  1. import gzip
  2. import json
  4. def write_compressed_json(data, filename="system_metrics.json.gz"):
  5.     with gzip.open(filename, 'wt', encoding='utf-8') as file:
  6.         json.dump(data, file, indent=4)

四、实际应用建议

  1. 采样间隔选择：

    • 短间隔（1-5秒）：适合实时监控
    • 长间隔（60秒+）：适合长期趋势分析

  2. 数据轮转策略：

    1. import os
    3. def rotate_log(filename, max_size_mb=10):
    4.     if os.path.exists(filename) and os.path.getsize(filename) > max_size_mb * 1024 * 1024:
    5.         backup_name = f"{filename}.{datetime.now().strftime('%Y%m%d%H%M%S')}"
    6.         os.rename(filename, backup_name)

  3. 可视化集成：

    • 使用pandas读取CSV文件进行数据分析
    • 结合matplotlib或Plotly生成可视化图表

五、性能优化技巧

  1. 减少I/O操作：

    • 批量写入代替频繁单次写入
    • 使用内存缓冲区暂存数据

  2. 多线程实现：

    1. import threading
    3. def async_write(data, writer_func):
    4.     thread = threading.Thread(target=writer_func, args=(data,))
    5.     thread.start()

  3. 选择高效存储格式：

    • 小数据量：JSON/CSV
    • 大数据量：Parquet/HDF5（需pandas支持）

六、完整代码示例

  1. import psutil
  2. import csv
  3. import json
  4. from datetime import datetime
  5. import time
  6. import gzip
  7. import os
  9. def get_cpu_usage():
  10.     return {
  11.         "total_usage": psutil.cpu_percent(interval=1),
  12.         "per_core_usage": psutil.cpu_percent(interval=1, percpu=True)
  13.     }
  15. def get_memory_info():
  16.     mem = psutil.virtual_memory()
  17.     return {
  18.         "total": mem.total,
  19.         "available": mem.available,
  20.         "used": mem.used,
  21.         "percent": mem.percent
  22.     }
  24. def get_disk_info():
  25.     disk_usage = psutil.disk_usage('/')
  26.     io_before = psutil.disk_io_counters()
  27.     time.sleep(1)
  28.     io_after = psutil.disk_io_counters()
  30.     read_speed = (io_after.read_bytes - io_before.read_bytes) / 1024
  31.     write_speed = (io_after.write_bytes - io_before.write_bytes) / 1024
  33.     return {
  34.         "usage": {
  35.             "total": disk_usage.total,
  36.             "used": disk_usage.used,
  37.             "free": disk_usage.free,
  38.             "percent": disk_usage.percent
  39.         },
  40.         "io_speed": {
  41.             "read_kbps": read_speed,
  42.             "write_kbps": write_speed
  43.         }
  44.     }
  46. def write_to_csv(data, filename="system_metrics.csv"):
  47.     timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
  48.     flat_data = {
  49.         "timestamp": timestamp,
  50.         "cpu_total": data["cpu"]["total_usage"],
  51.         "cpu_core1": data["cpu"]["per_core_usage"][0] if len(data["cpu"]["per_core_usage"]) > 0 else 0,
  52.         "mem_used_percent": data["memory"]["percent"],
  53.         "disk_used_percent": data["disk"]["usage"]["percent"],
  54.         "disk_read_kbps": data["disk"]["io_speed"]["read_kbps"],
  55.         "disk_write_kbps": data["disk"]["io_speed"]["write_kbps"]
  56.     }
  58.     fieldnames = ["timestamp", "cpu_total", "cpu_core1", "mem_used_percent", 
  59.                   "disk_used_percent", "disk_read_kbps", "disk_write_kbps"]
  61.     with open(filename, mode='a', newline='') as file:
  62.         writer = csv.DictWriter(file, fieldnames=fieldnames)
  63.         if file.tell() == 0:
  64.             writer.writeheader()
  65.         writer.writerow(flat_data)
  67. def write_compressed_json(data, filename="system_metrics.json.gz"):
  68.     timestamp = datetime.now().isoformat()
  69.     full_data = {
  70.         "timestamp": timestamp,
  71.         "metrics": data
  72.     }
  74.     try:
  75.         with gzip.open(filename, 'rt') as file:
  76.             existing_data = json.loads(file.read())
  77.     except (FileNotFoundError, json.JSONDecodeError):
  78.         existing_data = []
  80.     existing_data.append(full_data)
  82.     with gzip.open(filename, 'wt', encoding='utf-8') as file:
  83.         json.dump(existing_data, file, indent=4)
  85. def main(interval=5, output_format="csv"):
  86.     while True:
  87.         metrics = {
  88.             "cpu": get_cpu_usage(),
  89.             "memory": get_memory_info(),
  90.             "disk": get_disk_info()
  91.         }
  93.         if output_format == "csv":
  94.             write_to_csv(metrics)
  95.         elif output_format == "json_gz":
  96.             write_compressed_json(metrics)
  98.         print(f"[{datetime.now()}] Metrics collected ({output_format})")
  99.         time.sleep(interval)
  101. if __name__ == "__main__":
  102.     # 使用示例：python script.py 5 csv
  103.     import sys
  104.     interval = int(sys.argv[1]) if len(sys.argv) > 1 else 5
  105.     output_format = sys.argv[2] if len(sys.argv) > 2 else "csv"
  106.     main(interval, output_format)

七、总结与展望

本文系统阐述了Python实现系统性能监控的核心方法，涵盖CPU、内存、磁盘等关键指标的采集技术，以及CSV/JSON两种主流存储方案的实现细节。通过完整的代码示例和优化建议，开发者可以快速构建定制化的监控系统。

未来发展方向包括：

  1. 集成Prometheus/Grafana构建可视化监控平台
  2. 添加网络性能监控（带宽、延迟）
  3. 实现异常检测与自动告警机制
  4. 支持容器化部署（Docker/K8s）

建议开发者根据实际需求选择合适的存储格式和采样间隔，并考虑添加数据压缩和轮转策略以应对长期监控场景。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数