使用Python+psutil开发一个简易系统监控工具
引言
在运维自动化、服务器巡检、故障排查和资源告警场景中,经常需要快速获取机器的 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} 未运行")
十、总结
psutil 是 Python 运维自动化中非常实用的系统监控库。它屏蔽了不同操作系统之间的差异,让我们可以用统一的 Python 代码获取 CPU、内存、磁盘和进程信息。
本文完成了一个简易系统监控工具,核心能力包括:
- 使用
psutil.cpu_percent()获取 CPU 使用率 - 使用
psutil.virtual_memory()获取内存状态 - 使用
psutil.disk_usage()和psutil.disk_partitions()获取磁盘空间 - 使用
psutil.process_iter()获取进程信息 - 使用循环和清屏实现实时刷新
- 整合为一个完整可运行的系统监控脚本
对于 Python 运维自动化学习者来说,这类脚本非常适合作为练手项目。它既能帮助理解系统资源指标,也能为后续开发巡检工具、告警工具和监控 Agent 打下基础。
以上就是使用Python+psutil开发一个简易系统监控工具的详细内容,更多关于Python psutil系统监控工具的资料请关注脚本之家其它相关文章!


最新评论