Python脚本如何监控同步任务资源占用

wen python案例 2026-06-25 254

本文目录导读：

使用psutil监控系统资源
监控特定目录的IO活动
结合系统资源监控的完整示例
Web界面监控（可选）
安装依赖
使用建议

我来介绍几种监控同步任务资源占用的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()