Python脚本分析CPU使用情况

引言

CPU使用率是衡量计算机系统性能的重要指标之一，它反映了处理器在特定时间段内的繁忙程度。通过监控和分析CPU使用情况，系统管理员和开发人员可以识别性能瓶颈、优化资源分配并预防潜在的系统问题。

技术背景

现代操作系统提供了多种方式来获取CPU使用率信息：
/proc/stat文件（Linux系统）

WMI（Windows Management Instrumentation）

跨平台的psutil库

系统命令行工具（如top、vmstat、mpstat等）

Python因其简洁的语法和丰富的库支持，成为实现CPU监控和分析的理想选择。

应用使用场景
系统性能监控：实时跟踪CPU使用率，识别异常峰值

资源优化：分析应用程序的CPU消耗模式

容量规划：基于历史数据预测未来资源需求

故障排查：诊断系统卡顿或响应缓慢的原因

自动化运维：设置阈值触发警报或自动扩展

不同场景下详细代码实现

场景1：实时CPU监控

import psutil
import time

def monitor_cpu(interval=1):
“”“实时监控CPU使用率”""
print(“CPU使用率监控 (按Ctrl+C退出)”)
try:
while True:
# 获取所有逻辑CPU核心的使用率
cpu_percent = psutil.cpu_percent(interval=interval, percpu=True)
avg_percent = sum(cpu_percent) / len(cpu_percent)

        # 格式化输出
        cores_info = " | ".join([f"Core {i}: {p}%" for i, p in enumerate(cpu_percent)])
        print(f"\r平均: {avg_percent:.1f}% | {cores_info}", end="")
except KeyboardInterrupt:
    print("\n监控结束")

if name == “main”:
monitor_cpu()

场景2：历史数据记录与分析

import psutil
import time
import csv
from datetime import datetime

def record_cpu_usage(duration=60, interval=1, output_file=“cpu_usage.csv”):
“”“记录CPU使用率到CSV文件”""
end_time = time.time() + duration
print(f"开始记录CPU使用率，持续{duration}秒…")

with open(output_file, mode='w', newline='') as file:
    writer = csv.writer(file)
    writer.writerow(['Timestamp', 'CPU Usage (%)', 'CPU Count', 'Load Average'])
    
    while time.time() < end_time:
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        cpu_percent = psutil.cpu_percent(interval=interval)
        cpu_count = psutil.cpu_count()
        load_avg = psutil.getloadavg()[0] if hasattr(psutil, 'getloadavg') else 'N/A'
        
        writer.writerow([timestamp, cpu_percent, cpu_count, load_avg])
        print(f"\r当前CPU使用率: {cpu_percent}%", end="")

print(f"\n记录完成，数据已保存到{output_file}")

if name == “main”:
record_cpu_usage(duration=300) # 记录5分钟数据

场景3：进程级CPU分析

import psutil
import time
from collections import defaultdict

def analyze_process_cpu(top_n=10, interval=1, duration=60):
“”“分析各进程的CPU使用情况”""
print(f"开始分析进程CPU使用情况，持续{duration}秒…")

process_stats = defaultdict(list)
end_time = time.time() + duration

while time.time() < end_time:
    for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']):
        try:
            process_stats[proc.info['name']].append(proc.info['cpu_percent'])
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    
    time.sleep(interval)

# 计算平均CPU使用率
avg_usage = {name: sum(stats)/len(stats) for name, stats in process_stats.items()}

# 按CPU使用率排序并取前N个
top_processes = sorted(avg_usage.items(), key=lambda x: x[1], reverse=True)[:top_n]

print("\nCPU消耗最高的进程:")
print("{:<30} {:<10}".format('进程名', '平均CPU使用率(%)'))
for name, usage in top_processes:
    print("{:<30} {:<10.1f}".format(name, usage))

if name == “main”:
analyze_process_cpu()

原理解释

CPU使用率计算原理

CPU使用率通常表示为在特定时间间隔内CPU非空闲时间占总时间的百分比。计算过程如下：
从操作系统获取CPU时间统计信息（通常来自/proc/stat）

计算总CPU时间 = user + nice + system + idle + iowait + irq + softirq + steal + guest

计算非空闲时间 = 总CPU时间 - idle时间

CPU使用率 = (非空闲时间差) / (总时间差) × 100%

psutil库的工作原理

psutil是一个跨平台库，它在不同操作系统上使用不同的底层API：
Linux: 读取/proc/stat和/proc/[pid]/stat

Windows: 使用Windows API和WMI

macOS: 使用sysctl和sysinfo调用

核心特性
跨平台支持：兼容Windows、Linux、macOS等主流操作系统

多粒度监控：支持系统级、核心级、进程级监控

历史数据分析：可记录和可视化长期趋势

低开销：轻量级实现，对系统性能影响小

实时性：可配置采样间隔，满足不同实时性需求

原理流程图

±------------------+ ±------------------+ ±------------------+
操作系统统计接口 --> Python监控脚本 --> 数据分析与可视化
(procfs/WMI/API) (psutil/自定义逻辑) (CSV/图表/警报)
±------------------+ ±------------------+ ±------------------+

v

±------------------+ ±------------------+
硬件CPU核心 用户/管理员

±------------------+ ±------------------+

操作系统通过内核接口收集CPU时间统计

Python脚本通过psutil或原生接口获取这些统计

脚本计算CPU使用率并处理数据

结果被记录、分析或触发警报

用户查看报告并采取相应措施

环境准备

硬件要求
任何支持Python的计算机

建议至少1GHz CPU和512MB内存

软件要求
Python 3.6+

必需库：psutil

可选库：pandas（数据分析）、matplotlib（可视化）

安装步骤

创建虚拟环境（可选）

python -m venv cpu_monitor
source cpu_monitor/bin/activate # Linux/macOS
cpu_monitor\Scripts\activate # Windows

安装依赖

pip install psutil

可选安装数据分析库

pip install pandas matplotlib

实际详细应用代码示例实现

综合监控与分析工具

import psutil
import time
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime

class CPUMonitor:
def init(self):
self.cpu_count = psutil.cpu_count()
self.cpu_logical = psutil.cpu_count(logical=True)
self.cpu_freq = psutil.cpu_freq()

def get_cpu_info(self):
    """获取CPU基本信息"""
    info = {
        "物理核心数": self.cpu_count,
        "逻辑核心数": self.cpu_logical,
        "当前频率(MHz)": self.cpu_freq.current,
        "最小频率(MHz)": self.cpu_freq.min,
        "最大频率(MHz)": self.cpu_freq.max

return info

def monitor(self, interval=1, duration=60):
    """监控CPU使用率并生成报告"""
    timestamps = []
    cpu_percents = []
    per_cpu_percents = []
    
    end_time = time.time() + duration
    print(f"开始监控CPU使用率，持续{duration}秒...")
    
    while time.time() < end_time:
        # 获取数据
        timestamp = datetime.now()
        cpu_percent = psutil.cpu_percent(interval=interval)
        percpu_percent = psutil.cpu_percent(interval=interval, percpu=True)
        
        # 记录数据
        timestamps.append(timestamp)
        cpu_percents.append(cpu_percent)
        per_cpu_percents.append(percpu_percent)
        
        # 实时显示
        print(f"\r{timestamp.strftime('%H:%M:%S')} - CPU使用率: {cpu_percent}%", end="")
    
    # 创建DataFrame
    df = pd.DataFrame({
        'Timestamp': timestamps,
        'CPU_Usage': cpu_percents
    })
    
    # 添加每个核心的使用率
    for i in range(self.cpu_logical):
        df[f'Core_{i}'] = [x[i] for x in per_cpu_percents]
    
    return df

def analyze(self, df):
    """分析CPU使用数据并生成可视化"""
    print("\n\n分析结果:")
    print(df.describe())
    
    # 设置图形大小
    plt.figure(figsize=(12, 6))
    
    # 绘制总CPU使用率
    plt.subplot(2, 1, 1)
    plt.plot(df['Timestamp'], df['CPU_Usage'], label='总CPU使用率')
    plt.title('CPU使用率趋势')
    plt.ylabel('使用率 (%)')
    plt.grid(True)
    plt.legend()
    
    # 绘制各核心使用率
    plt.subplot(2, 1, 2)
    for i in range(self.cpu_logical):
        plt.plot(df['Timestamp'], df[f'Core_{i}'], label=f'Core {i}')
    plt.xlabel('时间')
    plt.ylabel('使用率 (%)')
    plt.grid(True)
    plt.legend()
    
    plt.tight_layout()
    plt.show()
    
    return df.describe()

if name == “main”:
monitor = CPUMonitor()
print(“CPU基本信息:”)
for k, v in monitor.get_cpu_info().items():
print(f"{k}: {v}")

data = monitor.monitor(duration=30)  # 监控30秒
stats = monitor.analyze(data)

运行结果示例

CPU基本信息:
物理核心数: 4
逻辑核心数: 8
当前频率(MHz): 2900.0
最小频率(MHz): 800.0
最大频率(MHz): 4000.0

开始监控CPU使用率，持续30秒…
14:25:30 - CPU使用率: 12.5%

分析结果:
CPU_Usage Core_0 Core_1 Core_2 Core_3 Core_4 Core_5 Core_6 Core_7
count 30.000000 30.000000 30.000000 30.000000 30.000000 30.000000 30.000000 30.000000 30.000000
mean 15.233333 8.500000 12.066667 18.233333 20.100000 10.833333 14.366667 17.533333 15.366667
std 7.146028 8.073365 9.377618 9.300518 11.123141 7.760103 8.270823 9.108678 8.073365
min 5.200000 0.900000 1.800000 5.000000 5.100000 1.700000 3.700000 5.000000 4.900000
25% 9.425000 2.825000 4.225000 10.925000 11.025000 4.425000 7.725000 10.025000 8.825000
50% 14.450000 6.450000 9.450000 17.450000 18.450000 8.450000 13.450000 16.450000 14.450000
75% 19.600000 12.600000 18.600000 24.600000 27.600000 15.600000 19.600000 23.600000 20.600000
max 28.900000 26.900000 30.900000 35.900000 39.900000 25.900000 29.900000 33.900000 30.900000

同时会显示CPU使用率随时间变化的折线图，包括总使用率和各核心使用率。

测试步骤以及详细代码

测试1：CPU压力测试与监控

import multiprocessing
import time

def cpu_stress():
“”“模拟CPU密集型任务”""
while True:
[x*x for x in range(10000)]

def test_under_load(cpu_monitor, stress_workers=None):
“”"
在CPU负载下测试监控功能
:param cpu_monitor: CPUMonitor实例
:param stress_workers: 要创建的负载进程数，默认为逻辑核心数
“”"
if stress_workers is None:
stress_workers = multiprocessing.cpu_count()

print(f"创建{stress_workers}个负载进程模拟CPU压力...")
processes = []
for _ in range(stress_workers):

= multiprocessing.Process(target=cpu_stress)

    p.start()
    processes.append(p)

try:
    # 监控30秒
    data = cpu_monitor.monitor(duration=30)
    stats = cpu_monitor.analyze(data)
    
    # 检查是否检测到高负载
    max_usage = stats.loc['max']['CPU_Usage']
    if max_usage < 80:
        print("警告: 未检测到预期的高CPU负载")
    else:
        print("测试通过: 成功检测到高CPU负载")
        
    return data
finally:
    # 清理负载进程
    for p in processes:
        p.terminate()
    for p in processes:
        p.join()

if name == “main”:
monitor = CPUMonitor()
test_under_load(monitor, stress_workers=2) # 使用2个负载进程测试

测试2：空闲状态监控测试

def test_idle(cpu_monitor):
“”“测试系统空闲状态下的监控”""
print(“测试系统空闲状态…”)

# 确保系统空闲几秒
time.sleep(5)

# 监控10秒
data = cpu_monitor.monitor(duration=10)
stats = cpu_monitor.analyze(data)

# 检查空闲状态
max_usage = stats.loc['max']['CPU_Usage']
if max_usage > 20:
    print(f"警告: 空闲状态下检测到异常高CPU使用率 ({max_usage}%)")
else:
    print(f"测试通过: 空闲状态CPU使用率正常 (最高{max_usage}%)")

return data

if name == “main”:
monitor = CPUMonitor()
test_idle(monitor)

部署场景
本地开发环境部署

开发环境监控脚本 (dev_monitor.py)

import psutil
import logging
from datetime import datetime

class DevCPUMonitor:
def init(self, threshold=80):
self.threshold = threshold
logging.basicConfig(
filename=‘cpu_monitor.log’,
level=loggingINFO,
format=’%(asctime)s - %(levelname)s - %(message)s’
)

def run(self, interval=10):
    """开发环境持续监控"""
    logging.info("启动开发环境CPU监控")
    try:
        while True:
            usage = psutil.cpu_percent(interval=interval)
            if usage > self.threshold:
                msg = f"警告: 高CPU使用率 ({usage}%)"
                logging.warning(msg)
                print(msg)
            else:
                logging.info(f"当前CPU使用率: {usage}%")
    except KeyboardInterrupt:
        logging.info("监控正常停止")

if name == “main”:
monitor = DevCPUMonitor(threshold=85) # 设置阈值为85%
monitor.run()

生产环境部署（Docker容器）

Dockerfile

FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY cpu_monitor.py .

设置健康检查

HEALTHCHECK --interval=30s --timeout=3s
CMD python -c “import psutil; exit(0 if psutil.cpu_percent() < 90 else 1)”

CMD [“python”, “cpu_monitor.py”]

生产环境监控脚本 (cpu_monitor.py)

import psutil
import time
import json
from http.server import BaseHTTPRequestHandler, HTTPServer

class CPUHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == ‘/metrics’:
cpu_percent = psutil.cpu_percent(interval=1)
cpu_count = psutil.cpu_count()
load_avg = psutil.getloadavg()[0] if hasattr(psutil, ‘getloadavg’) else None

        response = {
            'cpu_percent': cpu_percent,
            'cpu_count': cpu_count,
            'load_avg': load_avg,
            'timestamp': time.time()

self.send_response(200)

        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode())
    else:
        self.send_response(404)
        self.end_headers()

def run_server(port=8080):
server = HTTPServer((‘0.0.0.0’, port), CPUHandler)
print(f"启动CPU监控服务，端口 {port}")
server.serve_forever()

if name == “main”:
run_server()

部署命令:
docker build -t cpu-monitor .
docker run -d -p 8080:8080 --name cpu-monitor cpu-monitor

疑难解答

常见问题1：权限不足

问题描述：在Linux上获取某些进程信息时出现psutil.AccessDenied错误

解决方案：
使用root权限运行脚本

或者设置适当的sudo权限：

  sudo visudo

添加以下行（替换username为实际用户名）

username ALL=(ALL) NOPASSWD: /usr/bin/python /path/to/script.py

常见问题2：Windows上的WMI错误

问题描述：在Windows上出现WMI相关错误

解决方案：
确保WMI服务正在运行（Windows Management Instrumentation）

重建WMI存储库：

  winmgmt /verifyrepository  # 检查WMI存储库

winmgmt /salvagerepository # 修复WMI存储库

或者使用性能计数器替代：

  import psutil

psutil.cpu_percent(interval=1, use_perf_counter=True)

常见问题3：监控间隔不准确

问题描述：实际监控间隔与设置的interval参数不符

解决方案：
考虑系统调用的开销，实际间隔=interval+获取数据的时间

对于精确间隔，使用独立线程：

  import threading

class PreciseMonitor:
def init(self, interval=1):
self.interval = interval
self._running = False

   def _monitor_loop(self):
       while self._running:
           start = time.time()
           # 获取CPU数据
           data = psutil.cpu_percent(interval=None)
           # 处理数据...
           
           # 精确等待
           elapsed = time.time() - start
           sleep_time = self.interval - elapsed
           if sleep_time > 0:
               time.sleep(sleep_time)
               
   def start(self):
       self._running = True
       thread = threading.Thread(target=self._monitor_loop)
       thread.start()
       
   def stop(self):
       self._running = False

未来展望
AI驱动的异常检测：集成机器学习算法自动识别异常CPU使用模式

云原生监控：增强对Kubernetes和容器化环境的支持

能耗监控：结合RAPL（Running Average Power Limit）接口监控CPU能耗

预测分析：基于时间序列预测未来CPU负载

边缘计算支持：优化资源受限设备上的监控效率

技术趋势与挑战

趋势
多架构支持：随着ARM架构的普及，需要更好的异构CPU监控

实时分析：流式处理技术的应用实现更实时的监控

可观测性集成：与日志、追踪等可观测性工具深度集成

Serverless监控：适应无服务器架构的短暂进程监控

挑战
监控开销：在极高频率监控时自身造成的系统负载

安全性：监控工具可能成为攻击者获取系统信息的途径

隐私合规：在监控员工设备时的隐私法律问题

虚拟化环境：准确监控虚拟机和容器中的CPU使用率

总结

Python凭借其丰富的生态系统和psutil等强大库，为CPU使用情况分析提供了高效灵活的解决方案。本文介绍了从基础监控到高级分析的完整实现，涵盖了多种实际应用场景。随着计算架构的演进和云原生技术的发展，CPU监控工具需要不断适应新的环境和需求。未来的发展方向将更加注重智能化、集成化和低开销，同时解决安全与隐私方面的挑战。

通过合理利用这些监控技术，组织可以显著提升系统稳定性、优化资源利用率并快速定位性能问题，为业务系统的高效运行提供有力保障。