Python实时监控案例如何实现？

wen python案例 2026-06-07 22:32:06 1

Python实时监控案例如何实现？从零搭建高效数据监控系统

目录导读

实时监控的核心概念
什么是实时监控？为什么Python适合做监控？
技术选型与工具对比
轮询 vs 事件驱动：哪种方式更优？
实战案例一：文件系统变化监控
- 使用watchdog监听目录新增、修改、删除
实战案例二：Web服务健康监控
- 基于requests+schedule实现HTTP状态检测
实战案例三：系统资源（CPU/内存/磁盘）实时追踪
- 结合psutil与matplotlib动态绘图
性能优化与异常处理
避免死循环、内存泄漏、通知风暴
问答环节
常见问题与解决思路

实时监控的核心概念

Q：实时监控就是不停轮询吗？
A：不一定，轮询（Polling）是最直接的方式，但高频轮询会造成资源浪费，现代实时监控更倾向于事件驱动——系统主动通知变化，而非被动等待检查，Python因其异步编程能力（asyncio）和丰富的第三方库，是实现轻量级监控的理想语言。

核心要素：

数据采集：定时/触发式采样
阈值判断：超出预设值时触发报警
响应动作：记录日志、发送邮件、调用API

技术选型与工具对比

场景	推荐库	特点
文件系统变化	watchdog	跨平台，基于inotify/ReadDirectoryChangesW
定时任务	schedule / APScheduler	简单CRON语法，支持持久化
HTTP检测	requests + async	轻量，支持超时与重试
系统资源	psutil	跨平台，CPU/内存/磁盘/网络全覆盖
数据可视化	matplotlib / Plotly	实时刷新图表，适合Dashboard
异步并发	asyncio / aiohttp	处理大量连接时性能优势明显

选型原则：中小项目用watchdog+schedule，大规模分布式场景可引入Celery或Redis队列。

实战案例一：文件系统变化监控

需求描述

监控/data/uploads目录，当有新文件写入或修改时，自动触发后续处理（如转码、备份）。

代码实现（使用watchdog）

import sys
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"[修改] {event.src_path}")
    def on_created(self, event):
        print(f"[新增] {event.src_path}")
if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else "."
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        time.sleep(60)  # 持续监控60秒
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

运行命令：

python monitor_dir.py /data/uploads

注意事项：

递归监控需设置recursive=True
高负载下可增加batch_delay参数合并事件

实战案例二：Web服务健康监控

需求描述

每分钟检查https://api.example.com/health是否返回200，若连续3次失败则发送报警邮件。

代码实现（requests + schedule）

import requests
import smtplib
from email.message import EmailMessage
import schedule
import time
FAIL_COUNT = 0
THRESHOLD = 3
def check_health():
    global FAIL_COUNT
    try:
        resp = requests.get("https://api.example.com/health", timeout=5)
        if resp.status_code == 200:
            FAIL_COUNT = 0
            print("[健康] 服务正常")
        else:
            raise Exception(f"状态码异常：{resp.status_code}")
    except Exception as e:
        FAIL_COUNT += 1
        print(f"[警告] 第{FAIL_COUNT}次失败：{e}")
        if FAIL_COUNT >= THRESHOLD:
            send_alert()
            FAIL_COUNT = 0
def send_alert():
    msg = EmailMessage()
    msg.set_content("API服务连续3次检查失败，请立即处理！")
    msg["Subject"] = "【紧急】服务监控报警"
    msg["From"] = "monitor@example.com"
    msg["To"] = "admin@example.com"
    with smtplib.SMTP("smtp.example.com", 587) as server:
        server.login("user", "pass")
        server.send_message(msg)
    print("[报警] 邮件已发送")
schedule.every(1).minutes.do(check_health)
while True:
    schedule.run_pending()
    time.sleep(10)

优化点：

使用retry库实现指数退避重试
异步发送邮件,避免阻塞主循环

实战案例三：系统资源实时追踪

需求描述

实时采集CPU使用率、内存占用和磁盘IO，并在控制台动态刷新图表。

代码实现（psutil + matplotlib动态绘图）

import psutil
import matplotlib.pyplot as plt
import matplotlib.animation as animation
from collections import deque
# 存储最近60个数据点 (每1秒采集一次，显示1分钟)
cpu_data = deque(maxlen=60)
mem_data = deque(maxlen=60)
def get_system_info():
    cpu_percent = psutil.cpu_percent(interval=1)
    mem_percent = psutil.virtual_memory().percent
    cpu_data.append(cpu_percent)
    mem_data.append(mem_percent)
    return cpu_percent, mem_percent
def animate(i):
    cpu, mem = get_system_info()
    ax1.clear()
    ax2.clear()
    ax1.plot(cpu_data, label="CPU %", color="red")
    ax1.set_ylim(0, 100)
    ax1.legend(loc="upper right")
    ax1.set_title("实时CPU监控")
    ax2.plot(mem_data, label="内存 %", color="blue")
    ax2.set_ylim(0, 100)
    ax2.legend(loc="upper right")
    ax2.set_title("实时内存监控")
    fig.suptitle(f"当前 CPU: {cpu}% | 内存: {mem}%")
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 6))
ani = animation.FuncAnimation(fig, animate, interval=1000)
plt.show()

扩展建议：

增加磁盘IO、网络流量监控
使用plotly生成网页版实时仪表板

性能优化与异常处理

常见陷阱及对策

问题	原因	解决方案
内存不断增长	未清理历史数据	使用`deque`固定长度或定期清理
报警风暴	短时间内大量报警	引入熔断机制（如1分钟内只发1次）
资源泄漏	文件句柄未释放	使用`with`上下文管理器
监控自身挂掉	未守护进程	使用`supervisor`或`systemd`托管
时间不同步	分布式机器时区不一致	统一使用UTC时间戳

关键代码片段：防止报警重复触发

from datetime import datetime, timedelta
last_alert_time = None
def send_alert_if_needed():
    global last_alert_time
    now = datetime.now()
    if last_alert_time and (now - last_alert_time) < timedelta(minutes=5):
        return  # 5分钟内不重复报警
    send_alert()
    last_alert_time = now

问答环节

Q1：watchdog 和 inotify 有什么区别？
A：inotify是Linux内核原生的文件事件通知机制；watchdog是Python封装库，底层自动选择inotify（Linux）、fsevents（Mac）或ReadDirectoryChangesW（Windows），实现跨平台兼容。

Q2：实时监控中如何保证数据一致性？
A：使用消息队列缓冲数据（如Redis List或Kafka），消费者从队列中处理，防止采集器与处理器速度不匹配导致丢数据。

Q3：能否监控十万级文件目录？
A：可以，但需注意：

避免递归全遍历,只监听根目录
使用watchdog的event_queue设置合适大小
配合asyncio异步处理事件

Q4：如何实现多级报警（警告/严重）？
A：定义多个阈值，CPU > 80% → 黄色警告，> 95% → 红色严重报警。

def classify_alert(cpu_usage):
    if cpu_usage >= 95:
        return "严重"
    elif cpu_usage >= 80:
        return "警告"
    return None

Q5：监控系统本身崩溃如何恢复？
A：使用看门狗进程（Watchdog Process），主监控子进程；或者将监控脚本注册为系统服务（Systemd Service），设置Restart=always。

通过以上三个真实案例（文件监控、Web健康检查、系统资源追踪），您已经掌握了Python实现实时监控的核心模式：事件监听 + 定时轮询 + 异步处理，实际生产环境中，建议将采集与存储分离（如存入InfluxDB），并用Grafana展示数据，所有代码均可直接运行（需提前安装对应库：pip install watchdog requests psutil matplotlib schedule），小步迭代，您也能搭建企业级监控系统。

标签：数据采集

本文地址： https://dfhcn.com/post/1797.html

文章来源： wen