Python脚本如何监控进程运行状态

wen python案例 31

本文目录导读:

Python脚本如何监控进程运行状态

  1. 使用 psutil 库(推荐)
  2. 获取详细进程信息
  3. 进程存活检测与自动重启
  4. 监控CPU和内存使用率
  5. 日志记录与告警

我来介绍几种在Python中监控进程运行状态的方法:

使用 psutil 库(推荐)

安装

pip install psutil

基础监控示例

import psutil
import time
import os
def monitor_process(process_name):
    """监控指定进程的运行状态"""
    while True:
        running = False
        for proc in psutil.process_iter(['pid', 'name', 'status']):
            try:
                if process_name.lower() in proc.info['name'].lower():
                    running = True
                    print(f"进程 {process_name} 正在运行 (PID: {proc.info['pid']})")
                    print(f"状态: {proc.info['status']}")
                    print(f"CPU使用率: {proc.cpu_percent()}%")
                    print(f"内存使用: {proc.memory_info().rss / 1024 / 1024:.2f} MB")
                    print("-" * 50)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        if not running:
            print(f"进程 {process_name} 未运行")
        time.sleep(5)  # 每5秒检查一次
# 使用示例
monitor_process("chrome")  # 监控Chrome浏览器

获取详细进程信息

import psutil
from datetime import datetime
def get_process_details(pid):
    """获取指定PID的进程详细信息"""
    try:
        proc = psutil.Process(pid)
        return {
            'pid': proc.pid,
            'name': proc.name(),
            'status': proc.status(),
            'cpu_percent': proc.cpu_percent(),
            'memory_percent': proc.memory_percent(),
            'memory_rss': proc.memory_info().rss / 1024 / 1024,  # MB
            'create_time': datetime.fromtimestamp(proc.create_time()),
            'num_threads': proc.num_threads(),
            'exe_path': proc.exe(),
            'cmdline': ' '.join(proc.cmdline())
        }
    except psutil.NoSuchProcess:
        return None
    except psutil.AccessDenied:
        return None
# 使用示例
if __name__ == "__main__":
    # 查找特定进程
    for proc in psutil.process_iter(['pid', 'name']):
        if "notepad" in proc.info['name'].lower():
            details = get_process_details(proc.info['pid'])
            if details:
                for key, value in details.items():
                    print(f"{key}: {value}")

进程存活检测与自动重启

import psutil
import subprocess
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class ProcessMonitor:
    def __init__(self, process_name, restart_command=None):
        self.process_name = process_name
        self.restart_command = restart_command
        self.running = False
    def is_running(self):
        """检查进程是否在运行"""
        for proc in psutil.process_iter(['name']):
            try:
                if self.process_name.lower() in proc.info['name'].lower():
                    self.running = True
                    return True
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        self.running = False
        return False
    def restart_process(self):
        """重启进程"""
        if self.restart_command:
            try:
                subprocess.Popen(self.restart_command, shell=True)
                logging.info(f"启动进程: {self.restart_command}")
                return True
            except Exception as e:
                logging.error(f"启动进程失败: {e}")
                return False
        return False
    def monitor(self, check_interval=5):
        """持续监控进程状态"""
        logging.info(f"开始监控进程: {self.process_name}")
        while True:
            if not self.is_running():
                logging.warning(f"进程 {self.process_name} 已停止")
                if self.restart_process():
                    logging.info("等待进程启动...")
                    time.sleep(3)
                    if self.is_running():
                        logging.info("进程已成功启动")
            else:
                if not self.running:
                    logging.info(f"进程 {self.process_name} 运行正常")
                    self.running = True
            time.sleep(check_interval)
# 使用示例
if __name__ == "__main__":
    # 监控notepad.exe,如果停止则重新启动
    monitor = ProcessMonitor(
        process_name="notepad",
        restart_command="notepad"
    )
    monitor.monitor(check_interval=5)

监控CPU和内存使用率

import psutil
import time
from collections import deque
class ResourceMonitor:
    def __init__(self, process_name, history_size=10):
        self.process_name = process_name
        self.cpu_history = deque(maxlen=history_size)
        self.memory_history = deque(maxlen=history_size)
    def get_process_resources(self):
        """获取进程资源使用情况"""
        for proc in psutil.process_iter(['pid', 'name']):
            try:
                if self.process_name.lower() in proc.info['name'].lower():
                    p = psutil.Process(proc.info['pid'])
                    cpu = p.cpu_percent(interval=0.1)
                    mem = p.memory_info().rss / 1024 / 1024
                    self.cpu_history.append(cpu)
                    self.memory_history.append(mem)
                    return {
                        'cpu_percent': cpu,
                        'memory_mb': mem,
                        'cpu_avg': sum(self.cpu_history) / len(self.cpu_history),
                        'memory_avg': sum(self.memory_history) / len(self.memory_history)
                    }
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        return None
    def monitor(self, interval=2):
        """持续监控资源使用"""
        while True:
            resources = self.get_process_resources()
            if resources:
                print(f"CPU: {resources['cpu_percent']:.1f}% (平均: {resources['cpu_avg']:.1f}%)")
                print(f"内存: {resources['memory_mb']:.1f} MB (平均: {resources['memory_avg']:.1f} MB)")
                print("-" * 40)
            else:
                print(f"未找到进程: {self.process_name}")
            time.sleep(interval)
# 使用示例
monitor = ResourceMonitor("python")
monitor.monitor(interval=3)

日志记录与告警

import psutil
import logging
import time
from datetime import datetime
class AdvancedMonitor:
    def __init__(self, config):
        self.config = config
        self.setup_logging()
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('process_monitor.log'),
                logging.StreamHandler()
            ]
        )
    def check_process(self, process_name):
        """检查进程状态并记录日志"""
        found = False
        for proc in psutil.process_iter(['pid', 'name', 'status']):
            try:
                if process_name.lower() in proc.info['name'].lower():
                    found = True
                    process = psutil.Process(proc.info['pid'])
                    cpu = process.cpu_percent()
                    memory = process.memory_percent()
                    if cpu > self.config['cpu_threshold']:
                        logging.warning(f"高CPU使用率: {process_name} ({cpu:.1f}%)")
                    if memory > self.config['memory_threshold']:
                        logging.warning(f"高内存使用率: {process_name} ({memory:.1f}%)")
                    return True
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        if not found:
            logging.error(f"进程未运行: {process_name}")
            if self.config.get('auto_restart'):
                self.restart_process(self.config['restart_command'])
        return found
    def restart_process(self, command):
        """重启进程"""
        try:
            import subprocess
            subprocess.Popen(command, shell=True)
            logging.info(f"执行重启命令: {command}")
        except Exception as e:
            logging.error(f"重启失败: {e}")
# 配置示例
config = {
    'processes': ['notepad', 'chrome'],
    'cpu_threshold': 50.0,
    'memory_threshold': 80.0,
    'check_interval': 10,
    'auto_restart': True,
    'restart_command': 'notepad'
}
# 使用示例
if __name__ == "__main__":
    monitor = AdvancedMonitor(config)
    while True:
        for process in config['processes']:
            monitor.check_process(process)
        time.sleep(config['check_interval'])
  1. psutil库 是最强大、最常用的进程监控库
  2. 可以实现进程存活检测、资源监控、自动重启等功能
  3. 注意错误处理(NoSuchProcess, AccessDenied)
  4. 合理设置监控间隔,避免过度消耗资源
  5. 添加日志记录便于问题追踪

根据你的具体需求选择合适的方案,最简单的场景可以直接使用psutil的基本功能。

抱歉,评论功能暂时关闭!