Python脚本如何统计同步任务耗时峰值

wen python案例 2026-06-25 254

本文目录导读：

基础统计方法
滑动窗口统计峰值
实时监控装饰器
使用示例
可视化峰值
高级峰值分析

基础统计方法

import time
import threading
from collections import deque
import numpy as np
class SyncTaskMonitor:
    def __init__(self):
        self.task_times = []  # 存储所有任务耗时
        self.lock = threading.Lock()
    def measure_task(self, task_func, *args, **kwargs):
        """测量单个任务耗时"""
        start_time = time.time()
        result = task_func(*args, **kwargs)
        end_time = time.time()
        duration = end_time - start_time
        with self.lock:
            self.task_times.append(duration)
        return result, duration
    def get_peak_stats(self):
        """获取峰值统计"""
        with self.lock:
            if not self.task_times:
                return None
            times = np.array(self.task_times)
            return {
                'max': np.max(times),
                'min': np.min(times),
                'average': np.mean(times),
                'median': np.median(times),
                'std': np.std(times),
                'p95': np.percentile(times, 95),
                'p99': np.percentile(times, 99),
                'count': len(times)
            }

滑动窗口统计峰值

class SlidingWindowPeakMonitor:
    def __init__(self, window_size=100):
        self.window = deque(maxlen=window_size)
        self.peak_values = []
        self.lock = threading.Lock()
    def add_measurement(self, duration):
        """添加新的耗时数据"""
        with self.lock:
            self.window.append(duration)
            # 计算当前窗口的峰值
            if len(self.window) > 1:
                current_peak = max(self.window)
                self.peak_values.append({
                    'timestamp': time.time(),
                    'peak': current_peak,
                    'window_avg': np.mean(self.window),
                    'current_duration': duration
                })
    def get_peak_summary(self):
        """获取峰值摘要"""
        with self.lock:
            if not self.peak_values:
                return None
            peaks = [p['peak'] for p in self.peak_values]
            return {
                'all_time_peak': max(peaks),
                'recent_peak': self.peak_values[-1]['peak'] if self.peak_values else 0,
                'average_peak': np.mean(peaks),
                'peak_times': len([p for p in peaks if p > np.mean(peaks) * 1.5])
            }

实时监控装饰器

import functools
import logging
def monitor_sync_task(name=None, threshold=5.0):
    """监控同步任务的装饰器"""
    monitor = SyncTaskMonitor()
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            task_name = name or func.__name__
            start_time = time.time()
            try:
                result = func(*args, **kwargs)
                duration = time.time() - start_time
                # 记录耗时
                monitor.task_times.append(duration)
                # 检查是否超过阈值
                if duration > threshold:
                    logging.warning(f"Task '{task_name}' took {duration:.2f}s (threshold: {threshold}s)")
                # 检查是否达到峰值
                current_peak = max(monitor.task_times[-50:]) if len(monitor.task_times) >= 50 else max(monitor.task_times)
                if duration >= current_peak * 0.9:
                    logging.info(f"Task '{task_name}' near peak: {duration:.2f}s")
                return result
            except Exception as e:
                duration = time.time() - start_time
                logging.error(f"Task '{task_name}' failed after {duration:.2f}s: {e}")
                raise
        wrapper.monitor = monitor
        return wrapper
    return decorator

使用示例

# 示例1：基本用法
monitor = SyncTaskMonitor()
# 模拟同步任务
def sync_task(task_id, delay=1):
    time.sleep(delay)
    return f"Task {task_id} completed"
# 执行任务并测量
for i in range(10):
    result, duration = monitor.measure_task(sync_task, i, delay=0.5 + np.random.random())
    print(f"Task {i}: {duration:.2f}s")
# 获取统计
stats = monitor.get_peak_stats()
print(f"峰值耗时: {stats['max']:.2f}s")
print(f"平均耗时: {stats['average']:.2f}s")
print(f"P95耗时: {stats['p95']:.2f}s")
# 示例2：使用装饰器
@monitor_sync_task(name="data_sync", threshold=3.0)
def data_sync():
    time.sleep(2 + np.random.random() * 2)
    return "Data synced"
# 多次执行
for _ in range(5):
    data_sync()
# 查看统计
peak_info = data_sync.monitor.get_peak_stats()
print(f"任务耗时峰值: {peak_info['max']:.2f}s")

可视化峰值

import matplotlib.pyplot as plt
def plot_peak_distribution(task_times):
    """绘制峰值分布"""
    times = np.array(task_times)
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))
    # 耗时分布
    ax1.hist(times, bins=30, alpha=0.7)
    ax1.axvline(np.max(times), color='r', linestyle='--', label=f'Peak: {np.max(times):.2f}s')
    ax1.axvline(np.mean(times), color='g', linestyle='--', label=f'Avg: {np.mean(times):.2f}s')
    ax1.set_xlabel('Duration (s)')
    ax1.set_ylabel('Frequency')
    ax1.legend()
    ax1.set_title('Task Duration Distribution')
    # 时间序列
    ax2.plot(times, 'b-', alpha=0.6)
    ax2.axhline(np.max(times), color='r', linestyle='--', label='Peak')
    ax2.axhline(np.mean(times), color='g', linestyle='--', label='Average')
    ax2.set_xlabel('Task Number')
    ax2.set_ylabel('Duration (s)')
    ax2.legend()
    ax2.set_title('Task Duration Timeline')
    plt.tight_layout()
    plt.show()

高级峰值分析

class AdvancedPeakAnalyzer:
    def __init__(self, data):
        self.data = np.array(data)
    def find_peak_windows(self, window_size=10):
        """找出峰值窗口"""
        peaks = []
        for i in range(len(self.data) - window_size + 1):
            window = self.data[i:i+window_size]
            peaks.append({
                'start_idx': i,
                'end_idx': i + window_size,
                'peak': np.max(window),
                'mean': np.mean(window)
            })
        return peaks
    def detect_anomalies(self, threshold=2):
        """检测异常峰值"""
        mean = np.mean(self.data)
        std = np.std(self.data)
        anomalies = []
        for i, val in enumerate(self.data):
            if abs(val - mean) > threshold * std:
                anomalies.append({
                    'index': i,
                    'value': val,
                    'z_score': (val - mean) / std
                })
        return anomalies

这些方法可以帮助你：