本文目录导读：
我来介绍几种监控同步任务资源占用的Python方法：
使用psutil监控系统资源
import psutil
import time
import os
from datetime import datetime
class SyncMonitor:
def __init__(self, process_name=None, pid=None):
self.process_name = process_name
self.pid = pid
self.process = None
def find_process(self):
"""查找监控的进程"""
if self.pid:
try:
self.process = psutil.Process(self.pid)
return True
except psutil.NoSuchProcess:
return False
if self.process_name:
for proc in psutil.process_iter(['name', 'pid']):
if self.process_name in proc.info['name']:
self.process = proc
self.pid = proc.info['pid']
return True
return False
def get_resource_usage(self):
"""获取资源使用情况"""
if not self.process:
return None
try:
cpu_percent = self.process.cpu_percent(interval=1)
memory_info = self.process.memory_info()
io_counters = self.process.io_counters()
return {
'timestamp': datetime.now().isoformat(),
'pid': self.pid,
'cpu_percent': cpu_percent,
'memory_rss': memory_info.rss / 1024 / 1024, # MB
'memory_vms': memory_info.vms / 1024 / 1024, # MB
'io_read_bytes': io_counters.read_bytes / 1024 / 1024, # MB
'io_write_bytes': io_counters.write_bytes / 1024 / 1024, # MB
'num_threads': self.process.num_threads(),
'num_fds': self.process.num_fds()
}
except (psutil.NoSuchProcess, psutil.AccessDenied):
return None
def monitor_continuously(self, interval=5, duration=None):
"""持续监控"""
start_time = time.time()
records = []
while True:
if duration and (time.time() - start_time) > duration:
break
resource_data = self.get_resource_usage()
if resource_data:
records.append(resource_data)
print(f"[{resource_data['timestamp']}] "
f"CPU: {resource_data['cpu_percent']:.1f}% | "
f"内存: {resource_data['memory_rss']:.1f}MB | "
f"IO读: {resource_data['io_read_bytes']:.1f}MB | "
f"IO写: {resource_data['io_write_bytes']:.1f}MB")
time.sleep(interval)
return records
监控特定目录的IO活动
import threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class SyncActivityMonitor:
def __init__(self, watch_dirs):
self.watch_dirs = watch_dirs
self.observer = Observer()
self.activity_stats = {
'created': 0,
'modified': 0,
'deleted': 0,
'moved': 0,
'bytes_transferred': 0
}
class SyncEventHandler(FileSystemEventHandler):
def __init__(self, monitor):
self.monitor = monitor
def on_created(self, event):
self.monitor.activity_stats['created'] += 1
if not event.is_directory:
try:
self.monitor.activity_stats['bytes_transferred'] += \
event.src_path.stat().st_size
except:
pass
def on_modified(self, event):
self.monitor.activity_stats['modified'] += 1
def on_deleted(self, event):
self.monitor.activity_stats['deleted'] += 1
def on_moved(self, event):
self.monitor.activity_stats['moved'] += 1
def start_monitoring(self):
"""开始监控"""
handler = self.SyncEventHandler(self)
for watch_dir in self.watch_dirs:
self.observer.schedule(handler, watch_dir, recursive=True)
self.observer.start()
def stop_monitoring(self):
"""停止监控"""
self.observer.stop()
self.observer.join()
def get_statistics(self):
"""获取统计信息"""
return self.activity_stats.copy()
结合系统资源监控的完整示例
import subprocess
import json
import signal
import sys
class ComprehensiveSyncMonitor:
def __init__(self):
self.monitors = {}
self.is_running = False
def monitor_sync_process(self, sync_command, name="sync_task"):
"""监控特定的同步命令"""
try:
process = subprocess.Popen(
sync_command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
preexec_fn=os.setsid # 创建新会话，便于终止
)
self.monitors[name] = {
'process': process,
'cpu_monitor': SyncMonitor(pid=process.pid),
'start_time': time.time()
}
return process.pid
except Exception as e:
print(f"启动同步进程失败: {e}")
return None
def monitor_all_resources(self, interval=2):
"""监控所有同步任务的资源"""
self.is_running = True
while self.is_running:
for name, monitor in self.monitors.items():
process = monitor['process']
if process.poll() is not None:
print(f"同步任务 {name} 已结束")
continue
resource_data = monitor['cpu_monitor'].get_resource_usage()
if resource_data:
print(f"\n=== {name} 资源使用 ===")
print(f"CPU使用率: {resource_data['cpu_percent']:.1f}%")
print(f"内存使用: {resource_data['memory_rss']:.1f} MB")
print(f"线程数: {resource_data['num_threads']}")
# 检查资源是否异常
if resource_data['cpu_percent'] > 80:
print(f"警告: {name} CPU使用率过高!")
if resource_data['memory_rss'] > 500: # 500MB
print(f"警告: {name} 内存使用过高!")
time.sleep(interval)
def stop_all(self):
"""停止所有监控"""
self.is_running = False
for name, monitor in self.monitors.items():
process = monitor['process']
if process.poll() is None:
# 终止进程组
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
def log_statistics(self, duration=60):
"""记录统计信息到文件"""
import csv
start_time = time.time()
records = []
while (time.time() - start_time) < duration:
for name, monitor in self.monitors.items():
resource_data = monitor['cpu_monitor'].get_resource_usage()
if resource_data:
records.append({
**resource_data,
'task_name': name
})
time.sleep(5)
# 保存到CSV
if records:
filename = f"sync_monitor_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
with open(filename, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=records[0].keys())
writer.writeheader()
writer.writerows(records)
print(f"监控数据已保存到 {filename}")
# 使用示例
if __name__ == "__main__":
# 创建监控器
monitor = ComprehensiveSyncMonitor()
# 启动同步任务（示例：rsync）
pid = monitor.monitor_sync_process(
"rsync -avz /source/dir/ /destination/dir/",
"rsync_backup"
)
if pid:
print(f"同步任务已启动，PID: {pid}")
try:
# 持续监控30秒
monitor.log_statistics(duration=30)
except KeyboardInterrupt:
print("\n收到中断信号，停止监控...")
finally:
monitor.stop_all()
Web界面监控（可选）
from flask import Flask, jsonify
import threading
app = Flask(__name__)
monitor = SyncMonitor(process_name="rsync")
@app.route('/api/sync/status')
def get_sync_status():
resource_data = monitor.get_resource_usage()
if resource_data:
return jsonify(resource_data)
return jsonify({'error': '同步进程未运行'}), 404
@app.route('/api/sync/history')
def get_sync_history():
# 从数据库或文件中读取历史数据
pass
def run_web_server():
app.run(host='0.0.0.0', port=5000)
# 启动Web服务器
web_thread = threading.Thread(target=run_web_server, daemon=True)
web_thread.start()
安装依赖
pip install psutil watchdog flask
使用建议
- 定期记录：将资源使用情况记录到日志文件
- 告警机制：设置阈值，当资源使用异常时发送告警
- 图表展示：结合matplotlib或Grafana生成可视化图表
- 定期清理：定期清理历史监控数据
这个监控系统可以帮助您实时了解同步任务的资源使用情况,及时发现并解决性能问题。
