Python进程监控案例怎么编写？

wen python案例 2026-06-07 08:48:09 2

Python进程监控案例编写指南：从零实现高效守护脚本

目录导读

为什么需要进程监控？
核心监控逻辑设计
案例1：基于psutil的简易监控
案例2：结合日志与告警的进阶方案
案例3：守护进程自恢复框架
常见问题与问答（FAQ）

为什么需要进程监控？

在生产环境中，关键服务（如Web服务器、数据处理任务、爬虫脚本）一旦意外崩溃、卡死或资源泄漏，如果没有自动检测与恢复机制，会导致业务中断数小时。进程监控的核心目标包括：

实时检测进程是否存在（PID存活）
监控资源消耗（CPU/内存/文件句柄）
异常退出后自动重启
记录事件日志便于事后排查

问题1：为什么不用操作系统自带的systemd或supervisor？
答：对于复杂业务逻辑（如自定义健康检查、多进程分片管理），Python脚本可提供更灵活的决策逻辑,比如根据内存阈值降级服务。

核心监控逻辑设计

一个完备的Python进程监控方案通常包含以下模块：

[监控循环] → [进程状态检测] → [资源阈值判断] → [日志记录] → [告警/重启]

状态检测：通过os.kill(pid, 0)或psutil.pid_exists()确认进程存活
资源监控：psutil.Process(pid).memory_percent() 获取内存占比
健康检查：通过HTTP接口（如requests.get('http://localhost:8080/health')）判断服务响应
动作执行：subprocess.Popen(['python', 'app.py']) 重新启动进程

问题2：监控脚本本身崩溃怎么办？
答：使用crontab或systemd timer定时重启监控脚本，或结合watchdog库实现心跳自保。

案例1：基于psutil的简易监控

目标：监控主进程（PID已知）是否存在,若消失则重启。

import psutil
import subprocess
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
def monitor_process(pid, script_path):
    while True:
        if not psutil.pid_exists(pid):
            logging.warning(f"进程 {pid} 已消失，正在重启...")
            proc = subprocess.Popen(['python', script_path])
            pid = proc.pid
            logging.info(f"新进程 PID: {pid}")
        else:
            try:
                p = psutil.Process(pid)
                cpu_percent = p.cpu_percent(interval=1)
                mem_percent = p.memory_percent()
                logging.info(f"CPU: {cpu_percent:.1f}% / 内存: {mem_percent:.1f}%")
            except psutil.NoSuchProcess:
                continue
        time.sleep(10)
if __name__ == "__main__":
    # 初始启动目标进程
    target = subprocess.Popen(['python', 'my_service.py'])
    monitor_process(target.pid, 'my_service.py')

关键点：

使用psutil.pid_exists()比os.kill(pid, 0)更安全（不会抛出异常）
每10秒检查一次，避免高频轮询浪费资源
重启后需更新PID变量

案例2：结合日志与告警的进阶方案

当资源超过阈值时,除了重启还可以发送告警通知。

import smtplib
from email.mime.text import MIMEText
def send_alert(subject, body):
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = 'monitor@example.com'
    msg['To'] = 'admin@example.com'
    with smtplib.SMTP('smtp.example.com') as server:
        server.login('user', 'pass')
        server.send_message(msg)
def check_resource(pid):
    process = psutil.Process(pid)
    mem = process.memory_percent()
    if mem > 80:
        send_alert("内存超阈值", f"进程 {pid} 内存占用 {mem:.1f}%")
        process.terminate()  # 或直接重启
        return False
    return True

优势：

资源异常时自动终止进程并通知运维
可扩展至CPU、磁盘IO、网络连接数监控

问题3：如何监控多个进程？
答：维护一个进程列表pids = { 'app1': pid1, 'app2': pid2 },循环遍历每个PID执行相同逻辑。

案例3：守护进程自恢复框架

如果需要监控脚本自身不被kill,可设计父子进程心跳模式。

import os
import signal
def start_daemon():
    pid = os.fork()
    if pid == 0:  # 子进程（被监控目标）
        # 执行业务代码
        while True:
            time.sleep(1)
    else:         # 父进程（监控者）
        while True:
            if not psutil.pid_exists(pid):
                logging.error("子进程死亡，父进程退出并触发重启")
                exit(1)  # 由外部systemd重新拉起
            time.sleep(5)
if __name__ == "__main__":
    start_daemon()

适用场景：

对稳定性要求极高的核心服务，使用fork()创建隔离进程
父进程监控子进程，子进程死亡后父进程主动退出，触发上级守护

常见问题与问答（FAQ）

Q1：监控脚本占用CPU较高怎么办？

避免过短的时间间隔（建议≥5秒）
使用psutil的interval参数（如cpu_percent(interval=2)）而不是per_cpu=True
对于非实时场景，使用time.sleep()控制轮询粒度

Q2：进程状态检测的正确方式？

推荐psutil.pid_exists()，它处理了僵尸进程等边缘情况
subprocess.Popen.poll()也可用于子进程状态判断

Q3：如何防止监控脚本重复启动？

在脚本开头写PID文件：with open('/var/run/monitor.pid','w') as f: f.write(str(os.getpid()))
启动前检查文件中的PID是否存活

Q4：监控脚本需要root权限吗？

如果只监控当前用户进程，不需要root
若需监控系统进程或执行systemctl restart，则需要sudo权限

Q5：如何与外部告警系统集成？

通过HTTP请求发送到钉钉/飞书：requests.post('https://open.feishu.cn/open-apis/bot/v2/hook/xxxx', json={'msg_type':'text','content':{'text':'报警信息'}})
或集成prometheus_client做指标暴露，由监控集群采集

编写Python进程监控脚本的核心在于：明确监控目标（存活/资源/健康）、选择合适的检测库（psutil）、设计健壮的重启与告警逻辑，实际生产环境中，建议将监控脚本也加入systemd管理，并设置定时健康检查，通过本指南提供的三个案例,你可以快速搭建从基础存活检测到完整守护框架的监控系统。