本文目录导读:

我来介绍几种在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'])
- psutil库 是最强大、最常用的进程监控库
- 可以实现进程存活检测、资源监控、自动重启等功能
- 注意错误处理(NoSuchProcess, AccessDenied)
- 合理设置监控间隔,避免过度消耗资源
- 添加日志记录便于问题追踪
根据你的具体需求选择合适的方案,最简单的场景可以直接使用psutil的基本功能。