1024yysd 零一像素1024编程开发案例系统资源实时监控看板 -

系统资源实时监控看板

使用 Psutil 库获取当前操作系统的 CPU 使用率和内存占用情况。该脚本可以作为后台服务运行，用于服务器性能监控和健康度检查。

快速预览：

import psutil
import time
while True:
    cpu_percent = psutil.cpu_percent(interval=1)
    memory_info = psutil.virtual_memory()
    print(f"CPU: {cpu_percent}% | 内存: {memory_info.percent}%")
    time.sleep(2)

应用场景与价值

系统资源监控是运维和性能优化的基础工具，在以下场景中发挥关键作用：

服务器运维: 实时监控生产服务器的 CPU、内存、磁盘使用情况
性能调优: 识别资源瓶颈，优化应用程序性能
容量规划: 根据历史数据预测资源需求，合理分配计算资源
异常告警: 当资源使用率超过阈值时自动发送警报
开发调试: 监控程序运行时的资源消耗，发现内存泄漏等问题
系统健康检查: 定期检查系统状态，确保服务稳定运行

通过 Python 的 psutil 库，可以轻松获取系统各项指标，构建专业的监控系统，提前发现和解决性能问题。

基础知识

psutil 库

psutil (process and system utilities) 是一个跨平台的系统监控库，支持 Windows、Linux、macOS 等操作系统。

核心功能模块

1. CPU 监控

import psutil

# CPU 使用率（百分比）
psutil.cpu_percent(interval=1)  # 1秒内的平均使用率

# 每个核心的使用率
psutil.cpu_percent(interval=1, percpu=True)

# CPU 核心数
psutil.cpu_count()               # 逻辑核心数
psutil.cpu_count(logical=False)  # 物理核心数

# CPU 频率
psutil.cpu_freq()

2. 内存监控

# 物理内存
mem = psutil.virtual_memory()
mem.total      # 总内存（字节）
mem.available  # 可用内存
mem.used       # 已用内存
mem.percent    # 使用率（百分比）

# 交换内存（Swap）
swap = psutil.swap_memory()
swap.total
swap.used
swap.percent

3. 磁盘监控

# 磁盘分区
psutil.disk_partitions()

# 磁盘使用情况
disk = psutil.disk_usage('/')
disk.total
disk.used
disk.free
disk.percent

# 磁盘 I/O 统计
psutil.disk_io_counters()

4. 网络监控

# 网络 I/O 统计
net = psutil.net_io_counters()
net.bytes_sent
net.bytes_recv

# 网络连接
psutil.net_connections()

5. 进程管理

# 列出所有进程
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']):
    print(proc.info)

# 获取特定进程
proc = psutil.Process(pid)
proc.name()
proc.cpu_percent()
proc.memory_info()

代码详解

让我们深入分析系统监控代码的工作原理：

import psutil
import time

导入 psutil 用于系统监控，time 用于控制刷新间隔。

while True:

创建无限循环，持续监控系统资源。程序将一直运行，直到手动终止（Ctrl+C）。

    cpu_percent = psutil.cpu_percent(interval=1)

获取 CPU 使用率。interval=1 表示测量1秒内的平均使用率。第一次调用可能返回 0.0，需要一定时间采样。

    memory_info = psutil.virtual_memory()

获取内存信息对象，包含 total、available、percent、used 等属性。

    print(f"CPU: {cpu_percent}% | 内存: {memory_info.percent}%")

格式化输出监控数据。memory_info.percent 是已用内存的百分比。

    time.sleep(2)

暂停2秒后继续下一次监控。这控制了监控的刷新频率，避免过度占用 CPU。

完整代码

基础版本：实时监控看板

import psutil
import time
from datetime import datetime

def get_size(bytes):
    """
    将字节转换为人类可读的格式

    参数:
        bytes: 字节数

    返回:
        格式化的字符串，如 "1.5 GB"
    """
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if bytes < 1024:
            return f"{bytes:.2f} {unit}"
        bytes /= 1024
    return f"{bytes:.2f} PB"


def display_system_info():
    """显示系统监控信息"""
    print("\n" + "=" * 70)
    print(f"     系统资源监控看板 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 70)

    # CPU 信息
    cpu_percent = psutil.cpu_percent(interval=1)
    cpu_count = psutil.cpu_count()
    cpu_count_logical = psutil.cpu_count(logical=True)

    print(f"\n【CPU 信息】")
    print(f"  物理核心数: {cpu_count}")
    print(f"  逻辑核心数: {cpu_count_logical}")
    print(f"  总体使用率: {cpu_percent}%")

    # 每个核心的使用率
    cpu_per_core = psutil.cpu_percent(interval=1, percpu=True)
    for i, percent in enumerate(cpu_per_core):
        bar = '█' * int(percent / 5) + '░' * (20 - int(percent / 5))
        print(f"  核心 {i}: {bar} {percent}%")

    # 内存信息
    mem = psutil.virtual_memory()
    print(f"\n【内存信息】")
    print(f"  总内存: {get_size(mem.total)}")
    print(f"  已用内存: {get_size(mem.used)}")
    print(f"  可用内存: {get_size(mem.available)}")
    print(f"  使用率: {mem.percent}%")

    mem_bar = '█' * int(mem.percent / 5) + '░' * (20 - int(mem.percent / 5))
    print(f"  [{mem_bar}] {mem.percent}%")

    # 交换内存
    swap = psutil.swap_memory()
    if swap.total > 0:
        print(f"\n【交换内存】")
        print(f"  总交换空间: {get_size(swap.total)}")
        print(f"  已用: {get_size(swap.used)}")
        print(f"  使用率: {swap.percent}%")

    # 磁盘信息
    print(f"\n【磁盘信息】")
    partitions = psutil.disk_partitions()
    for partition in partitions:
        try:
            usage = psutil.disk_usage(partition.mountpoint)
            print(f"  设备: {partition.device}")
            print(f"  挂载点: {partition.mountpoint}")
            print(f"  总空间: {get_size(usage.total)}")
            print(f"  已用: {get_size(usage.used)}")
            print(f"  使用率: {usage.percent}%")
            print()
        except PermissionError:
            continue

    # 网络信息
    net = psutil.net_io_counters()
    print(f"【网络信息】")
    print(f"  发送: {get_size(net.bytes_sent)}")
    print(f"  接收: {get_size(net.bytes_recv)}")

    print("=" * 70)
    print("按 Ctrl+C 停止监控\n")


def monitor_loop(interval=5):
    """
    监控循环

    参数:
        interval: 刷新间隔（秒）
    """
    try:
        while True:
            # 清屏（可选，取消注释以启用）
            # import os
            # os.system('clear' if os.name == 'posix' else 'cls')

            display_system_info()
            time.sleep(interval)

    except KeyboardInterrupt:
        print("\n监控已停止")


# 运行监控
if __name__ == "__main__":
    monitor_loop(interval=3)

进阶版本：资源告警系统

import psutil
import time
from datetime import datetime
import logging

# 配置日志
logging.basicConfig(
    filename='system_monitor.log',
    level=logging.WARNING,
    format='%(asctime)s - %(levelname)s - %(message)s'
)


class SystemMonitor:
    """系统监控类"""

    def __init__(self, thresholds=None):
        """
        初始化监控器

        参数:
            thresholds: 告警阈值字典，如 {'cpu': 80, 'memory': 85, 'disk': 90}
        """
        self.thresholds = thresholds or {
            'cpu': 80,
            'memory': 85,
            'disk': 90
        }
        self.alert_history = []

    def check_cpu(self):
        """检查 CPU 使用率"""
        cpu_percent = psutil.cpu_percent(interval=1)

        if cpu_percent > self.thresholds['cpu']:
            message = f"⚠️  CPU 使用率过高: {cpu_percent}%"
            self.alert(message, 'WARNING')
            return False

        return True

    def check_memory(self):
        """检查内存使用率"""
        mem = psutil.virtual_memory()

        if mem.percent > self.thresholds['memory']:
            message = f"⚠️  内存使用率过高: {mem.percent}% (已用 {get_size(mem.used)} / {get_size(mem.total)})"
            self.alert(message, 'WARNING')
            return False

        return True

    def check_disk(self):
        """检查磁盘使用率"""
        issues = []

        for partition in psutil.disk_partitions():
            try:
                usage = psutil.disk_usage(partition.mountpoint)
                if usage.percent > self.thresholds['disk']:
                    message = f"⚠️  磁盘 {partition.mountpoint} 使用率过高: {usage.percent}%"
                    issues.append(message)
                    self.alert(message, 'WARNING')
            except PermissionError:
                continue

        return len(issues) == 0

    def check_processes(self, top_n=5):
        """检查资源占用最高的进程"""
        processes = []

        for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
            try:
                processes.append(proc.info)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass

        # 按 CPU 使用率排序
        top_cpu = sorted(processes, key=lambda x: x['cpu_percent'] or 0, reverse=True)[:top_n]

        # 按内存使用率排序
        top_mem = sorted(processes, key=lambda x: x['memory_percent'] or 0, reverse=True)[:top_n]

        return top_cpu, top_mem

    def alert(self, message, level='WARNING'):
        """
        发送告警

        参数:
            message: 告警消息
            level: 告警级别 (INFO, WARNING, ERROR)
        """
        # 记录到日志
        if level == 'WARNING':
            logging.warning(message)
        elif level == 'ERROR':
            logging.error(message)
        else:
            logging.info(message)

        # 记录到历史
        self.alert_history.append({
            'time': datetime.now(),
            'message': message,
            'level': level
        })

        # 显示告警
        print(f"[{datetime.now().strftime('%H:%M:%S')}] {message}")

        # 这里可以添加其他告警方式，如发送邮件、短信等
        # send_email_alert(message)

    def generate_report(self):
        """生成监控报告"""
        print("\n" + "=" * 70)
        print("     系统监控报告")
        print("=" * 70)

        # 系统概览
        cpu_percent = psutil.cpu_percent(interval=1)
        mem = psutil.virtual_memory()
        disk = psutil.disk_usage('/')

        print(f"\n【系统概览】")
        print(f"  CPU 使用率: {cpu_percent}% {'⚠️' if cpu_percent > self.thresholds['cpu'] else '✓'}")
        print(f"  内存使用率: {mem.percent}% {'⚠️' if mem.percent > self.thresholds['memory'] else '✓'}")
        print(f"  磁盘使用率: {disk.percent}% {'⚠️' if disk.percent > self.thresholds['disk'] else '✓'}")

        # 资源占用 TOP 5
        top_cpu, top_mem = self.check_processes(5)

        print(f"\n【CPU 占用 TOP 5】")
        for i, proc in enumerate(top_cpu, 1):
            print(f"  {i}. {proc['name']} (PID: {proc['pid']}) - {proc['cpu_percent']:.1f}%")

        print(f"\n【内存占用 TOP 5】")
        for i, proc in enumerate(top_mem, 1):
            print(f"  {i}. {proc['name']} (PID: {proc['pid']}) - {proc['memory_percent']:.1f}%")

        # 告警历史
        if self.alert_history:
            print(f"\n【告警历史】（最近10条）")
            for alert in self.alert_history[-10:]:
                print(f"  [{alert['time'].strftime('%H:%M:%S')}] {alert['level']}: {alert['message']}")
        else:
            print(f"\n【告警历史】")
            print("  ✅ 暂无告警")

        print("=" * 70 + "\n")

    def monitor(self, interval=5, report_interval=60):
        """
        持续监控

        参数:
            interval: 检查间隔（秒）
            report_interval: 报告生成间隔（秒）
        """
        print("系统监控已启动...\n")
        last_report_time = time.time()

        try:
            while True:
                # 执行检查
                self.check_cpu()
                self.check_memory()
                self.check_disk()

                # 定期生成报告
                if time.time() - last_report_time >= report_interval:
                    self.generate_report()
                    last_report_time = time.time()

                time.sleep(interval)

        except KeyboardInterrupt:
            print("\n监控已停止")
            self.generate_report()


# 使用示例
if __name__ == "__main__":
    monitor = SystemMonitor(thresholds={
        'cpu': 70,
        'memory': 80,
        'disk': 85
    })

    # 启动监控
    monitor.monitor(interval=5, report_interval=30)

高级版本：Web 可视化看板

import psutil
import time
from flask import Flask, render_template, jsonify
from datetime import datetime

app = Flask(__name__)

def get_system_data():
    """获取系统数据"""
    # CPU
    cpu_percent = psutil.cpu_percent(interval=1)
    cpu_per_core = psutil.cpu_percent(interval=1, percpu=True)

    # 内存
    mem = psutil.virtual_memory()

    # 磁盘
    disk = psutil.disk_usage('/')

    # 网络
    net = psutil.net_io_counters()

    return {
        'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        'cpu': {
            'total': cpu_percent,
            'per_core': cpu_per_core,
            'count': psutil.cpu_count()
        },
        'memory': {
            'total': mem.total,
            'available': mem.available,
            'used': mem.used,
            'percent': mem.percent
        },
        'disk': {
            'total': disk.total,
            'used': disk.used,
            'free': disk.free,
            'percent': disk.percent
        },
        'network': {
            'bytes_sent': net.bytes_sent,
            'bytes_recv': net.bytes_recv
        }
    }


@app.route('/')
def index():
    """主页"""
    return render_template('monitor.html')


@app.route('/api/system')
def api_system():
    """API 接口：返回系统数据"""
    return jsonify(get_system_data())


# HTML 模板（保存为 templates/monitor.html）
HTML_TEMPLATE = '''
<!DOCTYPE html>
<html>
<head>
    <title>系统监控看板</title>
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        .metric { display: inline-block; width: 200px; margin: 10px; padding: 15px; background: #f0f0f0; border-radius: 5px; }
        .metric h3 { margin: 0 0 10px 0; }
        .metric .value { font-size: 24px; font-weight: bold; }
        canvas { max-width: 600px; }
    </style>
</head>
<body>
    <h1>系统监控看板</h1>
    <div id="metrics"></div>
    <canvas id="cpuChart"></canvas>
    <canvas id="memoryChart"></canvas>

    <script>
        function updateMetrics() {
            fetch('/api/system')
                .then(response => response.json())
                .then(data => {
                    document.getElementById('metrics').innerHTML = `
                        <div class="metric">
                            <h3>CPU</h3>
                            <div class="value">${data.cpu.total}%</div>
                        </div>
                        <div class="metric">
                            <h3>内存</h3>
                            <div class="value">${data.memory.percent.toFixed(1)}%</div>
                        </div>
                        <div class="metric">
                            <h3>磁盘</h3>
                            <div class="value">${data.disk.percent.toFixed(1)}%</div>
                        </div>
                    `;
                });
        }

        setInterval(updateMetrics, 2000);
        updateMetrics();
    </script>
</body>
</html>
'''


if __name__ == '__main__':
    # 运行 Web 服务器
    print("系统监控看板已启动: http://localhost:5000")
    app.run(debug=True, port=5000)

运行示例

安装依赖

uv add psutil flask

执行监控

uv run main.py

预期输出

======================================================================
     系统资源监控看板 - 2026-01-09 15:30:45
======================================================================

【CPU 信息】
  物理核心数: 4
  逻辑核心数: 8
  总体使用率: 25.3%
  核心 0: ████████░░░░░░░░░░░░ 35.2%
  核心 1: ██████░░░░░░░░░░░░░░ 28.1%
  核心 2: ███████░░░░░░░░░░░░░ 32.5%
  核心 3: ████░░░░░░░░░░░░░░░░ 18.7%
  ...

【内存信息】
  总内存: 16.00 GB
  已用内存: 8.45 GB
  可用内存: 7.55 GB
  使用率: 52.8%
  [██████████░░░░░░░░░░] 52.8%

【磁盘信息】
  设备: /dev/disk1
  挂载点: /
  总空间: 500.00 GB
  已用: 320.50 GB
  使用率: 64.1%

【网络信息】
  发送: 25.30 GB
  接收: 102.45 GB

======================================================================
按 Ctrl+C 停止监控

扩展思路

1. 导出监控数据

import csv
from datetime import datetime

def export_to_csv(data, filename='system_metrics.csv'):
    """导出监控数据到 CSV"""
    with open(filename, 'a', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([
            datetime.now().isoformat(),
            data['cpu'],
            data['memory'],
            data['disk']
        ])

2. 集成 Prometheus

from prometheus_client import start_http_server, Gauge
import time

# 定义指标
cpu_gauge = Gauge('system_cpu_percent', 'CPU 使用率')
memory_gauge = Gauge('system_memory_percent', '内存使用率')

def collect_metrics():
    """收集指标供 Prometheus 抓取"""
    while True:
        cpu_gauge.set(psutil.cpu_percent())
        memory_gauge.set(psutil.virtual_memory().percent)
        time.sleep(5)

# 启动 Prometheus HTTP 服务器
start_http_server(8000)
collect_metrics()

3. 进程监控

def monitor_process(process_name):
    """监控特定进程"""
    for proc in psutil.process_iter(['pid', 'name']):
        if proc.info['name'] == process_name:
            print(f"进程 {process_name} (PID: {proc.info['pid']})")
            print(f"  CPU: {proc.cpu_percent()}%")
            print(f"  内存: {proc.memory_info().rss / 1024 / 1024:.2f} MB")

4. 性能趋势分析

from collections import deque

class TrendAnalyzer:
    def __init__(self, window_size=60):
        self.cpu_history = deque(maxlen=window_size)
        self.mem_history = deque(maxlen=window_size)

    def add_sample(self):
        self.cpu_history.append(psutil.cpu_percent())
        self.mem_history.append(psutil.virtual_memory().percent)

    def get_trend(self):
        """分析趋势（上升/下降/稳定）"""
        if len(self.cpu_history) < 10:
            return "数据不足"

        recent_avg = sum(list(self.cpu_history)[-10:]) / 10
        overall_avg = sum(self.cpu_history) / len(self.cpu_history)

        if recent_avg > overall_avg * 1.1:
            return "上升趋势"
        elif recent_avg < overall_avg * 0.9:
            return "下降趋势"
        else:
            return "稳定"

5. Docker 容器监控

import docker

client = docker.from_env()

def monitor_containers():
    """监控 Docker 容器资源使用"""
    for container in client.containers.list():
        stats = container.stats(stream=False)
        print(f"容器: {container.name}")
        print(f"  CPU: {stats['cpu_stats']['cpu_usage']['total_usage']}")
        print(f"  内存: {stats['memory_stats']['usage']}")

6. 系统健康评分

def calculate_health_score():
    """计算系统健康评分（0-100）"""
    cpu = psutil.cpu_percent()
    mem = psutil.virtual_memory().percent
    disk = psutil.disk_usage('/').percent

    # 加权评分
    score = 100
    score -= (cpu / 100) * 30  # CPU 权重 30%
    score -= (mem / 100) * 40  # 内存权重 40%
    score -= (disk / 100) * 30  # 磁盘权重 30%

    return max(0, score)

注意事项

性能影响

采样频率: 频繁调用监控函数会消耗 CPU 资源，建议间隔至少1秒
interval 参数: cpu_percent(interval=1) 会阻塞1秒，设计时注意
进程遍历: process_iter() 在进程很多时较慢，不宜频繁调用

权限问题

系统信息: 某些系统信息需要管理员权限
进程访问: 访问其他用户的进程可能被拒绝
跨平台: 不同操作系统的某些功能可能不可用

最佳实践

异常处理: 监控代码应有完善的异常处理，避免崩溃
日志记录: 记录监控数据和异常，便于事后分析
告警阈值: 根据实际情况设置合理的告警阈值
资源清理: 长期运行时注意内存管理，避免内存泄漏

通过掌握系统资源监控技术，你可以构建专业的运维工具，实时掌握系统状态，提前发现和解决性能问题，确保服务的稳定性和可靠性。