Python监控工具案例如何封装系统监控

wen python案例 2026-06-28 30

Python监控工具案例如何封装系统监控（附完整代码）

目录导读

为什么需要封装系统监控？从“写脚本”到“造工具”的思维转变
Python系统监控核心指标与数据采集方案
实战案例：基于psutil与logging的轻量级监控封装
监控数据持久化：从内存到数据库的优雅过渡
告警机制封装：如何让监控“主动说话”
扩展性设计：插件化架构让监控体系可维护
常见问题与问答

为什么需要封装系统监控？从“写脚本”到“造工具”的思维转变

在实际运维中，很多人习惯于写临时的监控脚本，比如每隔5秒打印一次CPU使用率，这种做法在单机临时调试时没问题，但一旦涉及多节点、长时间运行、数据可视化、告警策略等需求,原始脚本就会变得难以维护。

Python监控工具案例如何封装系统监控

封装的本质是什么？ 将零散的采集、解析、存储、告警逻辑抽象为可复用的模块，举个例子，下面的代码虽然能采集CPU信息,但无法复用：

import psutil
print(psutil.cpu_percent(interval=1))

而封装后的监控工具，应当支持配置化采集、多种输出方式、自定义阈值，这正是本文要探讨的核心：如何设计一套可扩展的系统监控封装方案。

Python系统监控核心指标与数据采集方案

系统监控主要围绕四大资源展开,每一项都包含多个细分指标：

资源类型	核心指标	采集工具/库
CPU	使用率、负载、上下文切换、中断	`psutil.cpu_percent()`
内存	总内存、已用、缓存、交换分区	`psutil.virtual_memory()`
磁盘	读写速率、IO等待时间、分区使用率	`psutil.disk_io_counters()`
网络	收发流量、连接数、丢包率	`psutil.net_io_counters()`

进阶监控还包括进程数、系统uptime、用户登录信息等，选择psutil作为核心采集库是因为它跨平台（Linux/Windows/macOS），且返回结构化的命名元组,便于后续封装。

实战案例：基于psutil与logging的轻量级监控封装

我们设计一个类SystemMonitor，核心思路是：功能内聚、接口统一、配置外置。

1 基础架构代码

import psutil
import time
import logging
from datetime import datetime
class SystemMonitor:
    def __init__(self, config=None):
        self.config = config or {}
        self.logger = self._setup_logger()
        self.metrics_history = []
    def _setup_logger(self):
        logger = logging.getLogger("SystemMonitor")
        handler = logging.StreamHandler()
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
        return logger
    def collect_cpu(self):
        return {
            "percent": psutil.cpu_percent(interval=1, percpu=False),
            "load_avg": psutil.getloadavg(),
            "times": psutil.cpu_times()._asdict() if hasattr(psutil.cpu_times(), '_asdict') else {}
        }
    def collect_memory(self):
        mem = psutil.virtual_memory()
        return {
            "total": mem.total,
            "available": mem.available,
            "percent": mem.percent,
            "used": mem.used
        }
    def collect_disk(self):
        disk = psutil.disk_usage('/')
        io = psutil.disk_io_counters()
        return {
            "usage": {"total": disk.total, "used": disk.used, "free": disk.free, "percent": disk.percent},
            "io": {"read_bytes": io.read_bytes, "write_bytes": io.write_bytes} if io else {}
        }
    def collect_all(self):
        return {
            "timestamp": datetime.now().isoformat(),
            "cpu": self.collect_cpu(),
            "memory": self.collect_memory(),
            "disk": self.collect_disk(),
            "network": self.collect_network()
        }
    def collect_network(self):
        net = psutil.net_io_counters()
        return {
            "bytes_sent": net.bytes_sent,
            "bytes_recv": net.bytes_recv,
            "packets_sent": net.packets_sent,
            "packets_recv": net.packets_recv
        }

2 数据采集的封装细节

采集粒度控制：通过interval参数控制CPU采样时间,避免短时波动导致误判。
结构化输出：统一返回字典格式，便于后续序列化（JSON）或写入数据库。
异常处理：采集磁盘或网络信息时，某些系统可能不支持所有指标，使用getattr或try-except兜底。

监控数据持久化：从内存到数据库的优雅过渡

数据采集后，需要存储以供分析,常见的方案包括：

方式	适用场景	优点	缺点
日志文件	单机调试	简单直接	查询不便
SQLite	小型项目	零配置	并发差
InfluxDB + Telegraf	生产环境	时序性能好	依赖外部服务
MySQL/PostgreSQL	需复杂查询	生态成熟	写入压力大

我们封装一个抽象存储接口,支持可拔插：

class BaseStorage:
    def save(self, metrics: dict):
        raise NotImplementedError
class SQLiteStorage(BaseStorage):
    def __init__(self, db_path="monitor.db"):
        import sqlite3
        self.conn = sqlite3.connect(db_path)
        self._create_table()
    def _create_table(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT,
                cpu_percent REAL,
                memory_percent REAL,
                disk_percent REAL
            )
        """)
        self.conn.commit()
    def save(self, metrics: dict):
        self.conn.execute(
            "INSERT INTO metrics (timestamp, cpu_percent, memory_percent, disk_percent) VALUES (?, ?, ?, ?)",
            (metrics["timestamp"], metrics["cpu"]["percent"], metrics["memory"]["percent"], metrics["disk"]["usage"]["percent"])
        )
        self.conn.commit()

这样，只需修改__init__中的存储对象,即可切换数据后端。

告警机制封装：如何让监控“主动说话”

告警是监控系统的灵魂,封装告警时需要考虑三个层次：

阈值定义：支持全局默认阈值和自定义每项指标的阈值。
触发条件：单次超阈值、连续N次超阈值（防抖动）、恢复通知。
通知渠道：邮件、飞书/钉钉Webhook、短信或自定义回调。

1 告警引擎封装示例

class AlertEngine:
    def __init__(self, threshold_config: dict, notifier=None):
        self.threshold = threshold_config  # {"cpu": 80, "memory": 90}
        self.notifier = notifier or DefaultNotifier()
        self.history = {}  # 记录连续超标次数
    def check(self, metrics: dict):
        alerts = []
        for key, value in self.threshold.items():
            current_val = self._extract_value(metrics, key)
            if current_val is None:
                continue
            if current_val > value:
                self.history[key] = self.history.get(key, 0) + 1
                if self.history[key] >= 3:  # 连续3次触发才告警
                    alerts.append(f"{key} 超标: {current_val}% (阈值 {value}%)")
                    self.history[key] = 0  # 重置计数
            else:
                self.history[key] = 0
        if alerts:
            self.notifier.send("\n".join(alerts))
        return alerts
    def _extract_value(self, metrics, key):
        # 支持多层路径，"cpu/percent"
        parts = key.split("/")
        val = metrics
        for part in parts:
            if isinstance(val, dict):
                val = val.get(part)
            else:
                return None
        return val

2 通知器接口

class BaseNotifier:
    def send(self, message: str):
        raise NotImplementedError
class ConsoleNotifier(BaseNotifier):
    def send(self, message):
        print(f"[ALERT] {message}")
class FeishuNotifier(BaseNotifier):
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    def send(self, message):
        import requests
        requests.post(self.webhook_url, json={"msg_type": "text", "content": {"text": message}})

扩展性设计：插件化架构让监控体系可维护

为了让系统监控真正“可封装”，我们引入插件机制，每个监控指标作为一个独立插件,实现统一接口：

class BasePlugin:
    name = ""
    def collect(self) -> dict: pass
    def alert_check(self, value) -> bool: pass
class CPUPlugin(BasePlugin):
    name = "cpu"
    def collect(self):
        return {"percent": psutil.cpu_percent()}
    def alert_check(self, value):
        return value > 80

主调度器通过插件列表动态加载：

class MonitorEngine:
    def __init__(self, plugins: list[BasePlugin]):
        self.plugins = plugins
        self.storage = SQLiteStorage()
        self.alert_engine = AlertEngine({"cpu": 80})
    def run_once(self):
        metrics = {"timestamp": datetime.now().isoformat()}
        for plugin in self.plugins:
            metrics[plugin.name] = plugin.collect()
        self.storage.save(metrics)
        self.alert_engine.check(metrics)

这样做的好处是：新增一个监控项（如GPU温度），只需编写一个新插件，无需修改任何现有代码。

常见问题与问答

Q1：系统监控封装中，最常踩的坑是什么？ A：主要有三点，第一是混淆“采集”与“监控”，采集只是第一步，真正封装需要包含数据处理、存储、告警链路，第二是跨平台兼容性，例如psutil在Linux下能获取loadavg，但在某些macOS版本下行为不同，建议代码中加入平台判断，第三是资源泄漏，长时间运行的监控程序要注意关闭数据库连接、文件句柄。

Q2：如何在不依赖第三方服务的前提下实现可视化？ A：可以使用Python内置的matplotlib生成实时图表，但更轻量的方式是输出JSON并通过静态HTTP服务器提供查看接口，或者集成Grafana + Prometheus，通过Python暴露/metrics端点,示例代码：

from prometheus_client import start_http_server, Gauge
cpu_gauge = Gauge('cpu_usage_percent', 'CPU usage in percent')
gpu_gauge = Gauge('gpu_usage_percent', 'GPU usage in percent')

Q3：封装时如何平衡通用性和复杂性？ A：遵循“配置驱动”原则，将可变的参数（采集频率、阈值、存储方式、通知渠道）全部外移至配置文件（YAML/JSON），核心代码保持稳定，当需要新增功能时，优先考虑在配置层扩展,而非修改封装层代码。

Q4：本文中的封装方案与市面上成熟的监控工具（如Nagios、Zabbix）有何区别？ A：本文方案侧重于“轻量级、快速定制、易于集成到现有Python项目”，对小型团队或特定业务场景（如监控GPU使用率、Web应用响应时间）非常友好，而Nagios等重型工具适合大型基础设施，但学习成本和运维负担更高，许多企业在使用Prometheus+Exporters模式,这与本文的插件化思想异曲同工。