本文目录导读：
我来帮您编写一个筛选慢速任务的脚本,这个脚本主要用于检测和筛选执行时间过长的任务。
通用慢速任务筛选脚本
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()
使用说明
-
选择适合的版本：
- Python版：适合大多数应用场景
- Node.js版：适合JavaScript生态
- Shell版：适合Linux系统监控
-
配置阈值：
- 根据实际需求调整时间/资源阈值
- 避免阈值过低导致误报
-
部署建议：
- 开发环境：配合测试使用
- 生产环境：设置适当的日志记录
- 性能要求高的场景：使用异步版本
这些脚本可以根据您的具体需求进行定制和扩展。
