使用Python+psutil开发一个简易系统监控工具_Python

引言

在运维自动化、服务器巡检、故障排查和资源告警场景中，经常需要快速获取机器的 cpu、内存、磁盘和进程状态。

如果只依赖系统命令，例如 linux 下的 top、free、df，或者 windows 下的任务管理器，自动化能力会比较弱。python 的 psutil 库正好解决了这个问题：它可以用统一的 python api 获取跨平台系统信息，非常适合用来开发轻量级系统监控脚本。

本文会从 psutil 基础用法开始，逐步实现一个可以实时刷新的简易系统监控工具。

一、psutil 库介绍

psutil 是 python 中常用的系统和进程监控库，全称可以理解为 process and system utilities。

它可以获取：

cpu 使用率、cpu 核心数、负载信息
内存总量、已用内存、可用内存、使用率
磁盘分区、磁盘容量、磁盘使用率
网络连接、网卡收发数据
当前运行的进程列表、进程 pid、进程名称、cpu 占用、内存占用

在运维自动化方向，psutil 常见用途包括：

编写服务器巡检脚本
定时采集系统资源指标
判断服务进程是否存活
查找高 cpu 或高内存进程
结合日志、邮件、企业微信或钉钉实现告警
作为监控 agent 的基础模块

安装方式很简单：

pip install psutil

验证安装：

import psutil

print(psutil.cpu_count())

如果能够输出 cpu 核心数量，说明安装成功。

二、获取 cpu 使用率

cpu 是系统监控中最核心的指标之一。使用 psutil.cpu_percent() 可以获取 cpu 使用率。

import psutil

cpu_percent = psutil.cpu_percent(interval=1)
print(f"cpu 使用率：{cpu_percent}%")

这里的 interval=1 表示统计 1 秒内的 cpu 使用情况。

也可以获取每个 cpu 核心的使用率：

import psutil

cpu_per_core = psutil.cpu_percent(interval=1, percpu=true)

for index, percent in enumerate(cpu_per_core, start=1):
    print(f"cpu 核心 {index}：{percent}%")

获取 cpu 核心数：

import psutil

logical_count = psutil.cpu_count(logical=true)
physical_count = psutil.cpu_count(logical=false)

print(f"逻辑核心数：{logical_count}")
print(f"物理核心数：{physical_count}")

在实际运维场景中，通常会重点关注整体 cpu 使用率。如果 cpu 长时间超过 80% 或 90%，就需要进一步排查是否存在异常进程、流量突增、死循环任务或定时任务集中运行等问题。

三、获取内存使用情况

内存监控可以使用 psutil.virtual_memory()。

import psutil

memory = psutil.virtual_memory()

print(f"内存总量：{memory.total}")
print(f"已用内存：{memory.used}")
print(f"可用内存：{memory.available}")
print(f"内存使用率：{memory.percent}%")

直接输出字节数不太方便阅读，可以封装一个单位转换函数：

def format_bytes(size):
    for unit in ["b", "kb", "mb", "gb", "tb"]:
        if size < 1024:
            return f"{size:.2f} {unit}"
        size /= 1024
    return f"{size:.2f} pb"

结合起来：

import psutil

def format_bytes(size):
    for unit in ["b", "kb", "mb", "gb", "tb"]:
        if size < 1024:
            return f"{size:.2f} {unit}"
        size /= 1024
    return f"{size:.2f} pb"

memory = psutil.virtual_memory()

print(f"内存总量：{format_bytes(memory.total)}")
print(f"已用内存：{format_bytes(memory.used)}")
print(f"可用内存：{format_bytes(memory.available)}")
print(f"内存使用率：{memory.percent}%")

对服务器来说，内存不足可能导致服务响应变慢、进程被系统终止、缓存命中率下降。运维脚本中一般会把内存使用率作为告警条件之一。

四、获取磁盘空间

磁盘空间可以通过 psutil.disk_usage() 获取。

import psutil

disk = psutil.disk_usage("/")

print(f"磁盘总量：{disk.total}")
print(f"已用空间：{disk.used}")
print(f"剩余空间：{disk.free}")
print(f"磁盘使用率：{disk.percent}%")

如果希望查看所有磁盘分区，可以使用 psutil.disk_partitions()：

import psutil

def format_bytes(size):
    for unit in ["b", "kb", "mb", "gb", "tb"]:
        if size < 1024:
            return f"{size:.2f} {unit}"
        size /= 1024
    return f"{size:.2f} pb"

partitions = psutil.disk_partitions()

for partition in partitions:
    try:
        usage = psutil.disk_usage(partition.mountpoint)
    except permissionerror:
        continue

    print(f"设备：{partition.device}")
    print(f"挂载点：{partition.mountpoint}")
    print(f"文件系统：{partition.fstype}")
    print(f"总空间：{format_bytes(usage.total)}")
    print(f"已用空间：{format_bytes(usage.used)}")
    print(f"剩余空间：{format_bytes(usage.free)}")
    print(f"使用率：{usage.percent}%")
    print("-" * 40)

磁盘监控非常重要。日志目录、数据库目录、上传文件目录如果没有清理策略，很容易出现磁盘被写满的问题。一旦磁盘满了，常见后果包括服务无法写日志、数据库写入失败、容器启动失败等。

五、获取进程信息

psutil 也可以获取系统中的进程信息。常用方法是 psutil.process_iter()。

import psutil

for process in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
    info = process.info
    print(
        f"pid={info['pid']}, "
        f"名称={info['name']}, "
        f"cpu={info['cpu_percent']}%, "
        f"内存={info['memory_percent']:.2f}%"
    )

在遍历进程时，可能遇到权限不足或进程已经退出的问题，所以正式脚本中通常要捕获异常：

import psutil

for process in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
    try:
        info = process.info
        print(info)
    except (psutil.nosuchprocess, psutil.accessdenied, psutil.zombieprocess):
        pass

如果要找出 cpu 占用较高的进程，可以排序：

import psutil
import time

# 第一次调用用于初始化 cpu 统计
for process in psutil.process_iter():
    try:
        process.cpu_percent(interval=none)
    except (psutil.nosuchprocess, psutil.accessdenied):
        pass

time.sleep(1)

processes = []
for process in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
    try:
        processes.append(process.info)
    except (psutil.nosuchprocess, psutil.accessdenied, psutil.zombieprocess):
        pass

top_processes = sorted(processes, key=lambda item: item["cpu_percent"], reverse=true)[:5]

for item in top_processes:
    print(
        f"pid={item['pid']}, "
        f"名称={item['name']}, "
        f"cpu={item['cpu_percent']}%, "
        f"内存={item['memory_percent']:.2f}%"
    )

这个功能在排查服务器负载过高时非常有用。可以快速定位是哪个 python 程序、java 服务、数据库进程或其他后台任务消耗了大量资源。

六、实时刷新系统状态

系统监控工具通常不是只执行一次，而是每隔几秒刷新一次状态。

基本思路如下：

清空终端屏幕
获取 cpu、内存、磁盘、进程信息
格式化输出
休眠指定秒数
循环执行

清屏函数可以兼容 windows、linux 和 macos：

import os

def clear_screen():
    os.system("cls" if os.name == "nt" else "clear")

循环刷新：

import time

while true:
    clear_screen()
    print("系统状态刷新中...")
    time.sleep(2)

为了方便停止程序，可以捕获 keyboardinterrupt。用户按 ctrl + c 时，脚本可以优雅退出。

try:
    while true:
        clear_screen()
        print("系统状态刷新中...")
        time.sleep(2)
except keyboardinterrupt:
    print("监控已停止")

七、制作一个简易系统监控脚本

下面实现一个完整脚本，功能包括：

显示当前时间
显示 cpu 核心数和 cpu 使用率
显示内存总量、可用内存、内存使用率
显示所有磁盘分区空间
显示 cpu 占用最高的前 5 个进程
每 2 秒自动刷新
支持 ctrl + c 停止

这个脚本适合用于入门级服务器巡检，也可以继续扩展为告警工具。

八、完整代码

文件名可以保存为 system_monitor.py。

import os
import time
from datetime import datetime

import psutil


refresh_interval = 2
top_process_limit = 5


def format_bytes(size):
    """convert bytes to a human-readable string."""
    for unit in ["b", "kb", "mb", "gb", "tb"]:
        if size < 1024:
            return f"{size:.2f} {unit}"
        size /= 1024
    return f"{size:.2f} pb"


def clear_screen():
    os.system("cls" if os.name == "nt" else "clear")


def get_cpu_info():
    return {
        "logical_count": psutil.cpu_count(logical=true),
        "physical_count": psutil.cpu_count(logical=false),
        "percent": psutil.cpu_percent(interval=1),
        "per_core": psutil.cpu_percent(interval=none, percpu=true),
    }


def get_memory_info():
    memory = psutil.virtual_memory()
    return {
        "total": memory.total,
        "available": memory.available,
        "used": memory.used,
        "percent": memory.percent,
    }


def get_disk_info():
    disks = []

    for partition in psutil.disk_partitions():
        try:
            usage = psutil.disk_usage(partition.mountpoint)
        except (permissionerror, filenotfounderror):
            continue

        disks.append(
            {
                "device": partition.device,
                "mountpoint": partition.mountpoint,
                "fstype": partition.fstype,
                "total": usage.total,
                "used": usage.used,
                "free": usage.free,
                "percent": usage.percent,
            }
        )

    return disks


def init_process_cpu_percent():
    for process in psutil.process_iter():
        try:
            process.cpu_percent(interval=none)
        except (psutil.nosuchprocess, psutil.accessdenied):
            continue


def get_top_processes(limit=5):
    processes = []

    for process in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
        try:
            info = process.info
            processes.append(
                {
                    "pid": info["pid"],
                    "name": info["name"] or "",
                    "cpu_percent": info["cpu_percent"] or 0.0,
                    "memory_percent": info["memory_percent"] or 0.0,
                }
            )
        except (psutil.nosuchprocess, psutil.accessdenied, psutil.zombieprocess):
            continue

    return sorted(processes, key=lambda item: item["cpu_percent"], reverse=true)[:limit]


def print_separator():
    print("-" * 80)


def print_system_status():
    now = datetime.now().strftime("%y-%m-%d %h:%m:%s")
    cpu = get_cpu_info()
    memory = get_memory_info()
    disks = get_disk_info()
    top_processes = get_top_processes(top_process_limit)

    print(f"python psutil 系统监控工具")
    print(f"刷新时间：{now}")
    print_separator()

    print("cpu 信息")
    print(f"物理核心数：{cpu['physical_count']}")
    print(f"逻辑核心数：{cpu['logical_count']}")
    print(f"cpu 总使用率：{cpu['percent']}%")
    print(f"各核心使用率：{', '.join(str(item) + '%' for item in cpu['per_core'])}")
    print_separator()

    print("内存信息")
    print(f"内存总量：{format_bytes(memory['total'])}")
    print(f"已用内存：{format_bytes(memory['used'])}")
    print(f"可用内存：{format_bytes(memory['available'])}")
    print(f"内存使用率：{memory['percent']}%")
    print_separator()

    print("磁盘信息")
    for disk in disks:
        print(
            f"{disk['device']} "
            f"挂载点={disk['mountpoint']} "
            f"文件系统={disk['fstype']} "
            f"总量={format_bytes(disk['total'])} "
            f"已用={format_bytes(disk['used'])} "
            f"剩余={format_bytes(disk['free'])} "
            f"使用率={disk['percent']}%"
        )
    print_separator()

    print(f"cpu 占用最高的前 {top_process_limit} 个进程")
    print(f"{'pid':<10}{'cpu%':<10}{'mem%':<10}进程名称")
    for process in top_processes:
        print(
            f"{process['pid']:<10}"
            f"{process['cpu_percent']:<10.1f}"
            f"{process['memory_percent']:<10.2f}"
            f"{process['name']}"
        )
    print_separator()
    print(f"每 {refresh_interval} 秒刷新一次，按 ctrl + c 停止监控")


def main():
    init_process_cpu_percent()

    try:
        while true:
            clear_screen()
            print_system_status()
            time.sleep(refresh_interval)
    except keyboardinterrupt:
        print("\n监控已停止")


if __name__ == "__main__":
    main()

运行脚本：

python system_monitor.py

如果是在 linux 服务器上，也可以使用：

python3 system_monitor.py

运行后，终端会每隔 2 秒刷新一次系统状态。

九、脚本扩展方向

这个脚本只是一个基础版本，后续可以继续扩展：

增加 cpu、内存、磁盘阈值判断
超过阈值后发送邮件、钉钉或企业微信告警
将监控数据写入 csv、sqlite、mysql 或 influxdb
增加网络流量监控
增加指定进程存活检测
使用 argparse 支持命令行参数，例如刷新间隔、进程数量、告警阈值
配合 schedule、cron 或 windows 任务计划程序实现定时巡检
封装成 flask/fastapi 接口，做成 web 监控面板

例如，增加一个简单的内存告警判断：

import psutil

memory = psutil.virtual_memory()

if memory.percent >= 80:
    print(f"警告：当前内存使用率过高，已达到 {memory.percent}%")

再比如，检查某个关键进程是否存在：

import psutil

target_process = "nginx"
exists = false

for process in psutil.process_iter(["name"]):
    try:
        if target_process.lower() in (process.info["name"] or "").lower():
            exists = true
            break
    except (psutil.nosuchprocess, psutil.accessdenied):
        continue

if not exists:
    print(f"警告：进程 {target_process} 未运行")