如何写筛选慢速任务脚本

wen 实用脚本 2026-06-26 106

本文目录导读：

如何写筛选慢速任务脚本

通用慢速任务筛选脚本
使用说明

我来帮您编写一个筛选慢速任务的脚本,这个脚本主要用于检测和筛选执行时间过长的任务。

通用慢速任务筛选脚本

Python版本（最通用）

#!/usr/bin/env python3
"""
慢速任务筛选脚本
用于检测执行时间超过阈值的任务
"""
import time
import threading
from datetime import datetime
from functools import wraps
class SlowTaskDetector:
    def __init__(self, threshold=1.0):
        self.threshold = threshold  # 慢速阈值（秒）
        self.slow_tasks = []
    def monitor_function(self, func):
        """装饰器：监控函数执行时间"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            result = func(*args, **kwargs)
            end_time = time.time()
            execution_time = end_time - start_time
            if execution_time > self.threshold:
                self.slow_tasks.append({
                    'function': func.__name__,
                    'args': args,
                    'kwargs': kwargs,
                    'execution_time': execution_time,
                    'timestamp': datetime.now()
                })
                print(f"⚠️ 慢速任务检测: {func.__name__} "
                      f"执行时间 {execution_time:.3f}秒")
            return result
        return wrapper
    def report(self):
        """生成慢速任务报告"""
        if not self.slow_tasks:
            print("✅ 没有检测到慢速任务")
            return
        print(f"\n📊 慢速任务报告（阈值: {self.threshold}秒）")
        print("=" * 50)
        for task in self.slow_tasks:
            print(f"函数: {task['function']}")
            print(f"执行时间: {task['execution_time']:.3f}秒")
            print(f"时间戳: {task['timestamp']}")
            print("-" * 50)
# 使用示例
detector = SlowTaskDetector(threshold=0.5)
@detector.monitor_function
def slow_function():
    """模拟慢速任务"""
    time.sleep(2)  # 模拟耗时操作
    return "完成"
@detector.monitor_function
def fast_function():
    """模拟快速任务"""
    time.sleep(0.1)  # 模拟快速操作
    return "完成"
if __name__ == "__main__":
    # 测试函数
    slow_function()
    fast_function()
    # 生成报告
    detector.report()

Bash/Shell版本（Linux系统）

#!/bin/bash
"""
慢速进程筛选脚本
用于找出消耗CPU时间过长的进程
"""
# 设置阈值（秒）
THRESHOLD=${1:-10}
echo "🔍 正在筛选运行时间超过 ${THRESHOLD}秒 的进程..."
echo "========================================"
# 使用ps命令筛选慢速进程
ps aux | awk -v threshold="$THRESHOLD" '
    NR>1 {
        # 获取进程运行时间（格式可能不同，这里简化处理）
        if ($10 ~ /:/) {
            split($10, time_parts, ":")
            minutes = time_parts[1]
            seconds = time_parts[2]
            total_seconds = minutes * 60 + seconds
        } else {
            total_seconds = $10
        }
        if (total_seconds > threshold) {
            printf "PID: %-8s CPU: %-5s MEM: %-5s TIME: %-10s COMMAND: %s\n", 
                   $2, $3, $4, $10, $11
        }
    }
'
# 高级版本：使用top命令
echo -e "\n📊 慢速进程TOP 10（按CPU使用率排序）"
echo "========================================"
ps aux --sort=-%cpu | head -11 | awk '
    NR>1 {
        printf "PID: %-8s CPU: %-5s%% MEM: %-5s%% COMMAND: %s\n", 
               $2, $3, $4, $11
    }
'

JavaScript/Node.js版本

/**
 * 慢速任务筛选脚本 (Node.js)
 */
class TaskMonitor {
    constructor(threshold = 1000) { // 阈值默认1000ms
        this.threshold = threshold;
        this.slowTasks = new Map();
        this.taskCount = 0;
    }
    // 监控异步函数
    monitor(fn, taskName = '') {
        return async (...args) => {
            const startTime = Date.now();
            const taskId = ++this.taskCount;
            const name = taskName || fn.name || `Task-${taskId}`;
            try {
                const result = await fn(...args);
                const endTime = Date.now();
                const duration = endTime - startTime;
                if (duration > this.threshold) {
                    this.recordSlowTask({
                        name,
                        duration,
                        args,
                        startTime: new Date(startTime),
                        endTime: new Date(endTime)
                    });
                }
                return result;
            } catch (error) {
                console.error(`任务 ${name} 执行失败:`, error);
                throw error;
            }
        };
    }
    // 记录慢速任务
    recordSlowTask(taskInfo) {
        console.warn(`⚠️ 慢速任务 [${taskInfo.name}]`, {
            duration: `${taskInfo.duration}ms`,
            threshold: `${this.threshold}ms`,
            time: taskInfo.startTime.toISOString()
        });
        this.slowTasks.set(taskInfo.name, taskInfo);
    }
    // 生成报告
    generateReport() {
        console.log('\n📊 慢速任务报告');
        console.log('='.repeat(50));
        if (this.slowTasks.size === 0) {
            console.log('✅ 没有检测到慢速任务');
            return;
        }
        this.slowTasks.forEach((task, name) => {
            console.log(`任务: ${name}`);
            console.log(`执行时间: ${task.duration}ms`);
            console.log(`开始时间: ${task.startTime.toISOString()}`);
            console.log('-'.repeat(50));
        });
    }
}
// 使用示例
const monitor = new TaskMonitor(500); // 500ms阈值
// 模拟慢速函数
const slowTask = monitor(async () => {
    await new Promise(resolve => setTimeout(resolve, 2000));
    return '慢速任务完成';
}, '数据库查询');
const fastTask = monitor(async () => {
    await new Promise(resolve => setTimeout(resolve, 100));
    return '快速任务完成';
}, '缓存读取');
// 执行测试
async function main() {
    await slowTask();
    await fastTask();
    monitor.generateReport();
}
main();

实时监控版本（持续运行）

#!/usr/bin/env python3
"""
实时慢速任务监控脚本
"""
import time
import psutil
import logging
from datetime import datetime
class RealTimeSlowTaskMonitor:
    def __init__(self, cpu_threshold=50, memory_threshold=100, check_interval=5):
        self.cpu_threshold = cpu_threshold
        self.memory_threshold = memory_threshold  # MB
        self.check_interval = check_interval
        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('slow_tasks.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def check_slow_processes(self):
        """检查慢速进程"""
        slow_processes = []
        for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_info', 'create_time']):
            try:
                # 计算运行时间
                create_time = proc.info['create_time']
                if create_time:
                    run_time = time.time() - create_time
                    # 检查CPU和内存使用
                    cpu_usage = proc.info['cpu_percent'] or 0
                    memory_usage = (proc.info['memory_info'].rss / 1024 / 1024) if proc.info['memory_info'] else 0
                    if cpu_usage > self.cpu_threshold or memory_usage > self.memory_threshold:
                        slow_processes.append({
                            'pid': proc.info['pid'],
                            'name': proc.info['name'],
                            'cpu': cpu_usage,
                            'memory': memory_usage,
                            'run_time': run_time
                        })
            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                pass
        return slow_processes
    def run(self):
        """启动监控"""
        self.logger.info(f"🔄 启动慢速进程监控")
        self.logger.info(f"阈值设置: CPU > {self.cpu_threshold}%, 内存 > {self.memory_threshold}MB")
        try:
            while True:
                slow_tasks = self.check_slow_processes()
                if slow_tasks:
                    self.logger.warning(f"⚠️ 发现 {len(slow_tasks)} 个慢速进程:")
                    for task in slow_tasks[:5]:  # 只显示前5个
                        self.logger.warning(
                            f"PID: {task['pid']}, "
                            f"名称: {task['name']}, "
                            f"CPU: {task['cpu']:.1f}%, "
                            f"内存: {task['memory']:.1f}MB, "
                            f"运行时间: {task['run_time']:.0f}秒"
                        )
                time.sleep(self.check_interval)
        except KeyboardInterrupt:
            self.logger.info("📊 监控已停止")
# 使用示例
if __name__ == "__main__":
    monitor = RealTimeSlowTaskMonitor(
        cpu_threshold=80,  # CPU超过80%视为慢速
        memory_threshold=200,  # 内存超过200MB视为慢速
        check_interval=10  # 每10秒检查一次
    )
    monitor.run()