🐍 第24课：系统监控

—— psutil：跨平台系统监控神器

🏆 CPU/内存/磁盘/进程全掌握

✅ Python验证通过

📌 本课目标

掌握 psutil 获取 CPU/内存/磁盘/网络信息
掌握进程级监控（查找/杀进程/资源占用）
掌握实时监控仪表盘脚本
掌握系统资源告警逻辑
掌握 psutil 与 SQLite 结合的监控日志

1️⃣ CPU 监控

import psutil

# CPU 使用率
cpu_percent = psutil.cpu_percent(interval=1)
print(f"CPU 使用率: {cpu_percent}%")

# 每核使用率
per_cpu = psutil.cpu_percent(interval=1, percpu=True)
for i, pct in enumerate(per_cpu):
    print(f"  核心{i}: {pct}%")

# CPU 信息
print(f"物理核心: {psutil.cpu_count(logical=False)}")
print(f"逻辑核心: {psutil.cpu_count(logical=True)}")
print(f"CPU 频率: {psutil.cpu_freq().current:.0f}MHz")

# CPU 时间统计
times = psutil.cpu_times()
print(f"用户态: {times.user:.1f}s")
print(f"系统态: {times.system:.1f}s")
print(f"空闲: {times.idle:.1f}s")

# CPU 统计（上下文切换/中断）
stats = psutil.cpu_stats()
print(f"上下文切换: {stats.ctx_switches}")
print(f"中断次数: {stats.interrupts}")

2️⃣ 内存监控

import psutil

# 物理内存
mem = psutil.virtual_memory()
print(f"总内存: {mem.total / 1024**3:.1f} GB")
print(f"可用: {mem.available / 1024**3:.1f} GB")
print(f"已用: {mem.used / 1024**3:.1f} GB")
print(f"使用率: {mem.percent}%")
print(f"缓存: {mem.cached / 1024**3:.1f} GB")

# 交换分区
swap = psutil.swap_memory()
print(f"\n交换分区:")
print(f"  总量: {swap.total / 1024**3:.1f} GB")
print(f"  已用: {swap.used / 1024**3:.1f} GB")
print(f"  使用率: {swap.percent}%")

# 格式化工具
def fmt_bytes(b):
    """格式化字节数"""
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if b < 1024:
            return f"{b:.1f} {unit}"
        b /= 1024
    return f"{b:.1f} PB"

print(f"\n内存: {fmt_bytes(mem.used)} / {fmt_bytes(mem.total)} ({mem.percent}%)")

3️⃣ 磁盘与网络

import psutil
from collections import Counter

# 磁盘分区
for part in psutil.disk_partitions():
    print(f"{part.device} → {part.mountpoint} ({part.fstype})")

# 磁盘使用率
for part in psutil.disk_partitions():
    try:
        usage = psutil.disk_usage(part.mountpoint)
        print(f"  {part.mountpoint}: {usage.percent}% ({usage.used//1024**3}G/{usage.total//1024**3}G)")
    except PermissionError:
        pass

# 磁盘 I/O
io = psutil.disk_io_counters()
print(f"\n磁盘I/O:")
print(f"  读: {io.read_bytes/1024**3:.1f}GB ({io.read_count}次)")
print(f"  写: {io.write_bytes/1024**3:.1f}GB ({io.write_count}次)")

# 网络 I/O
net = psutil.net_io_counters()
print(f"\n网络I/O:")
print(f"  发送: {net.bytes_sent/1024**3:.2f}GB")
print(f"  接收: {net.bytes_recv/1024**3:.2f}GB")

# 网络连接
connections = psutil.net_connections(kind="inet")
print(f"\n活跃连接: {len(connections)}")

# 按状态分组
states = Counter(c.status for c in connections)
for status, count in states.most_common():
    print(f"  {status}: {count}")

# 网卡信息
for name, addrs in psutil.net_if_addrs().items():
    for addr in addrs:
        if addr.family.name == "AF_INET":
            print(f"  {name}: {addr.address}")

4️⃣ 进程监控

import psutil

# 所有进程（按内存排序 Top10）
procs = []
for proc in psutil.process_iter(["pid", "name", "memory_percent", "cpu_percent"]):
    try:
        procs.append(proc.info)
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        pass

top10 = sorted(procs, key=lambda p: p["memory_percent"] or 0, reverse=True)[:10]
print("内存 Top10:")
for p in top10:
    print(f"  PID {p['pid']:>6} {p['name'][:20]:<20} MEM:{p['memory_percent']:5.1f}%")

# 单进程详情
def inspect_process(pid):
    """查看进程详情"""
    try:
        proc = psutil.Process(pid)
        print(f"PID: {proc.pid}")
        print(f"名称: {proc.name()}")
        print(f"状态: {proc.status()}")
        print(f"CPU: {proc.cpu_percent(interval=0.1):.1f}%")
        print(f"内存: {proc.memory_info().rss/1024**2:.1f}MB")
        print(f"线程数: {proc.num_threads()}")
        print(f"打开文件: {len(proc.open_files())}")
        print(f"连接数: {len(proc.connections())}")
    except psutil.NoSuchProcess:
        print(f"进程 {pid} 不存在")

# 按名称查找进程
def find_process(name):
    """按名称查找进程"""
    results = []
    for proc in psutil.process_iter(["pid", "name"]):
        if name.lower() in proc.info["name"].lower():
            results.append(proc.info)
    return results

5️⃣ 实时监控脚本

import psutil
import time

def monitor_dashboard(interval=2, count=5):
    """终端监控仪表盘"""
    for i in range(count):
        # 采集数据
        cpu = psutil.cpu_percent(interval=0)
        mem = psutil.virtual_memory()
        disk = psutil.disk_usage("/")
        
        # 进度条
        def bar(pct, width=30):
            filled = int(pct / 100 * width)
            return "█" * filled + "░" * (width - filled)
        
        # 输出
        print(f"\n{'='*50}")
        print(f"📊 系统监控 [{time.strftime('%H:%M:%S')}]")
        print(f"CPU  [{bar(cpu)}] {cpu:.1f}%")
        print(f"内存 [{bar(mem.percent)}] {mem.percent:.1f}% ({mem.used//1024**3}G/{mem.total//1024**3}G)")
        print(f"磁盘 [{bar(disk.percent)}] {disk.percent:.1f}% ({disk.used//1024**3}G/{disk.total//1024**3}G)")
        
        # 告警逻辑
        alerts = []
        if cpu > 80:
            alerts.append(f"⚠️ CPU使用率 {cpu:.1f}% 超过阈值!")
        if mem.percent > 85:
            alerts.append(f"⚠️ 内存使用率 {mem.percent:.1f}% 超过阈值!")
        if disk.percent > 90:
            alerts.append(f"⚠️ 磁盘使用率 {disk.percent:.1f}% 超过阈值!")
        
        if alerts:
            print("\n".join(alerts))
        else:
            print("✅ 系统状态正常")
        
        if i < count - 1:
            time.sleep(interval)

# 运行 3 次采样
monitor_dashboard(interval=1, count=3)

6️⃣ 验证脚本

#!/usr/bin/env python3
"""第24课 系统监控验证"""
import psutil

def test_cpu():
    cpu = psutil.cpu_percent(interval=0.1)
    assert 0 <= cpu <= 100
    logical = psutil.cpu_count(logical=True)
    assert logical >= 1
    physical = psutil.cpu_count(logical=False)
    assert physical >= 1
    print(f"✅ CPU: {cpu}% 物理{physical}核 逻辑{logical}核")

def test_memory():
    mem = psutil.virtual_memory()
    assert mem.total > 0
    assert 0 <= mem.percent <= 100
    assert mem.available <= mem.total
    print(f"✅ 内存: {mem.percent}% ({mem.total//1024**3}GB)")

def test_disk():
    partitions = psutil.disk_partitions()
    assert len(partitions) >= 1
    usage = psutil.disk_usage("/")
    assert usage.total > 0
    assert 0 <= usage.percent <= 100
    print(f"✅ 磁盘: {usage.percent}% ({usage.total//1024**3}GB)")

def test_network():
    net = psutil.net_io_counters()
    assert net.bytes_sent >= 0
    assert net.bytes_recv >= 0
    print(f"✅ 网络: 发送{net.bytes_sent//1024}KB 接收{net.bytes_recv//1024}KB")

def test_process():
    procs = list(psutil.process_iter(["pid", "name"]))
    assert len(procs) > 10
    current = psutil.Process()
    print(f"✅ 进程: 共{len(procs)}个, 当前PID={current.pid}")

if __name__ == "__main__":
    test_cpu()
    test_memory()
    test_disk()
    test_network()
    test_process()
    print("\n🎉 第24课全部验证通过！")

7️⃣ 监控数据记录

import psutil
import sqlite3
import time
from datetime import datetime

class MonitorDB:
    """监控数据持久化"""
    
    def __init__(self, db_path=":memory:"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""CREATE TABLE IF NOT EXISTS metrics (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            cpu_percent REAL,
            memory_percent REAL,
            disk_percent REAL
        )""")
        self.conn.commit()
    
    def record(self):
        cpu = psutil.cpu_percent(interval=0)
        mem = psutil.virtual_memory()
        disk = psutil.disk_usage("/")
        
        self.conn.execute(
            "INSERT INTO metrics (timestamp, cpu_percent, memory_percent, disk_percent) VALUES (?,?,?,?)",
            (datetime.now().isoformat(), cpu, mem.percent, disk.percent)
        )
        self.conn.commit()
        return {"cpu": cpu, "memory": mem.percent, "disk": disk.percent}
    
    def get_history(self, limit=100):
        rows = self.conn.execute(
            "SELECT * FROM metrics ORDER BY timestamp DESC LIMIT ?", (limit,)
        ).fetchall()
        return rows
    
    def close(self):
        self.conn.close()

# 使用
db = MonitorDB(":memory:")
for i in range(3):
    result = db.record()
    print(f"  采样{i+1}: CPU={result['cpu']}% MEM={result['memory']}%")
    time.sleep(1)

8️⃣ 进程管理工具

import psutil

class ProcessManager:
    """进程管理工具"""
    
    @staticmethod
    def find_by_name(name):
        """按名称查找进程"""
        return [p.info for p in psutil.process_iter(["pid", "name"])
                if name.lower() in p.info["name"].lower()]
    
    @staticmethod
    def find_by_port(port):
        """按端口查找进程"""
        for conn in psutil.net_connections(kind="inet"):
            if conn.laddr.port == port and conn.status == "LISTEN":
                try:
                    proc = psutil.Process(conn.pid)
                    return {"pid": conn.pid, "name": proc.name()}
                except psutil.NoSuchProcess:
                    pass
        return None
    
    @staticmethod
    def top_processes(sort_by="memory", n=10):
        """Top N 进程"""
        procs = []
        for p in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
            try:
                procs.append(p.info)
            except:
                pass
        key = sort_by if sort_by in ("cpu_percent", "memory_percent") else "memory_percent"
        return sorted(procs, key=lambda x: x.get(key) or 0, reverse=True)[:n]

# 使用示例
pm = ProcessManager()
top = pm.top_processes("memory", 5)
for p in top:
    print(f"  PID {p['pid']:>6} {p['name'][:15]:<15} MEM:{p['memory_percent']:.1f}%")

9️⃣ 系统信息汇总脚本

import psutil
import platform
import socket
from datetime import datetime

def system_report():
    """生成系统信息报告"""
    report = []
    report.append("=" * 50)
    report.append("🖥️ 系统信息报告")
    report.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    report.append("=" * 50)
    
    # 基本信息
    report.append(f"
📡 主机名: {socket.gethostname()}")
    report.append(f"操作系统: {platform.system()} {platform.release()}")
    report.append(f"架构: {platform.machine()}")
    report.append(f"Python: {platform.python_version()}")
    
    # CPU
    cpu_percent = psutil.cpu_percent(interval=1)
    report.append(f"
🔬 CPU:")
    report.append(f"  物理核心: {psutil.cpu_count(logical=False)}")
    report.append(f"  逻辑核心: {psutil.cpu_count(logical=True)}")
    report.append(f"  使用率: {cpu_percent}%")
    freq = psutil.cpu_freq()
    if freq:
        report.append(f"  频率: {freq.current:.0f}MHz")
    
    # 内存
    mem = psutil.virtual_memory()
    report.append(f"
💾 内存:")
    report.append(f"  总量: {mem.total / 1024**3:.1f} GB")
    report.append(f"  已用: {mem.used / 1024**3:.1f} GB ({mem.percent}%)")
    report.append(f"  可用: {mem.available / 1024**3:.1f} GB")
    
    swap = psutil.swap_memory()
    report.append(f"  交换: {swap.used / 1024**3:.1f} / {swap.total / 1024**3:.1f} GB")
    
    # 磁盘
    report.append(f"
💿 磁盘:")
    for part in psutil.disk_partitions():
        try:
            usage = psutil.disk_usage(part.mountpoint)
            report.append(f"  {part.mountpoint}: {usage.percent}% ({usage.used//1024**3}G/{usage.total//1024**3}G)")
        except PermissionError:
            pass
    
    # 网络
    net = psutil.net_io_counters()
    report.append(f"
🌐 网络:")
    report.append(f"  发送: {net.bytes_sent / 1024**3:.2f} GB")
    report.append(f"  接收: {net.bytes_recv / 1024**3:.2f} GB")
    
    # 进程
    procs = list(psutil.process_iter(["pid", "name", "memory_percent"]))
    report.append(f"
📊 进程: {len(procs)} 个")
    
    report.append("
" + "=" * 50)
    return "\n".join(report)

print(system_report())

🔑 本课要点

psutil——跨平台系统监控神器，CPU/内存/磁盘/网络一站搞定
cpu_percent()——interval 参数控制采样间隔
virtual_memory()——全面内存信息，available 比 free 更有用
Process()——单进程级别监控，内存/CPU/文件描述符
实时监控——while True + interval + 清屏 = 终端仪表盘

← 上一课 📚 目录下一课 →