本文目录导读：
基础统计方法
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
这些方法可以帮助你：
- 实时监控任务耗时
- 计算各种统计指标（最大值、P95、P99等）
- 检测异常峰值
- 可视化分析结果
- 设置告警阈值
根据你的具体需求选择合适的实现方式。
