HTTP 状态码：客户端与服务器的通信语言——第四部分：客户端错误状态码（4xx）深度解读（二）
第20章：405 Method Not Allowed - 方法不允许深度分析
20.1 定义与语义
405 Method Not Allowed 状态码表示请求行中指定的方法对请求URI标识的资源是已知的，但不被目标资源支持。服务器必须生成一个 Allow 响应头字段，包含目标资源当前支持的方法列表。
核心语义：
-
服务器理解请求，但明确拒绝执行该方法
-
与501（未实现）不同：405表示方法有效，但此资源不支持
-
与403（禁止）不同：405是关于HTTP方法，不是访问权限
关键要求：
-
必须包含
Allow头部 -
响应应该是客户端可以理解的
-
可以包含响应体解释错误原因
20.2 触发场景与分类
20.2.1 常见的405触发场景
python
# RESTful API中的典型405场景
class RESTfulAPI:
# 1. 对只读资源使用写方法
# GET /api/articles/123 ✅
# PUT /api/articles/123 ❌ 405
# 2. 对只写资源使用读方法
# POST /api/webhooks ✅
# GET /api/webhooks ❌ 405
# 3. 端点仅支持特定方法
# POST /api/auth/login ✅
# GET /api/auth/login ❌ 405
# 4. 版本化API中的方法弃用
# PATCH /api/v2/users/123 ✅
# PUT /api/v2/users/123 ❌ 405（PUT在v2中已弃用）
20.2.2 方法语义不匹配
http
# 示例：对集合使用不适当的方法
PUT /api/users HTTP/1.1
Content-Type: application/json
Content-Length: 45
{"name": "John", "email": "john@example.com"}
# 响应 - 应该使用POST创建资源
HTTP/1.1 405 Method Not Allowed
Allow: GET, POST, HEAD, OPTIONS
Content-Type: application/json
{
"error": {
"code": "METHOD_NOT_ALLOWED",
"message": "PUT method is not allowed on this collection. Use POST to create new resources.",
"allowed_methods": ["GET", "POST", "HEAD", "OPTIONS"],
"suggested_method": "POST"
}
}
20.3 详细实现与最佳实践
20.3.1 Allow头部的正确实现
python
# Flask实现：正确的Allow头部生成
from flask import Flask, request, jsonify
from functools import wraps
import inspect
class MethodAllowanceManager:
def __init__(self, app):
self.app = app
self.route_methods = {}
self.init_method_tracking()
def init_method_tracking(self):
"""初始化方法跟踪"""
@self.app.before_request
def track_route():
# 存储当前路由支持的方法
if request.endpoint:
rule = self.app.url_map._rules_by_endpoint[request.endpoint][0]
if request.endpoint not in self.route_methods:
self.route_methods[request.endpoint] = set(rule.methods)
def get_allowed_methods(self, endpoint):
"""获取端点允许的方法"""
if endpoint in self.route_methods:
return self.route_methods[endpoint]
# 如果没有找到，检查所有可能的路由
for rule in self.app.url_map.iter_rules():
if rule.endpoint == endpoint:
return set(rule.methods)
return set()
app = Flask(__name__)
method_manager = MethodAllowanceManager(app)
# 自定义405错误处理器
@app.errorhandler(405)
def handle_method_not_allowed(e):
"""生成正确的405响应"""
# 确定当前端点
endpoint = request.endpoint
# 获取允许的方法
allowed_methods = method_manager.get_allowed_methods(endpoint)
# 排序方法：常用方法在前
method_order = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS']
sorted_methods = sorted(
allowed_methods,
key=lambda x: method_order.index(x) if x in method_order else len(method_order)
)
# 生成响应
response = jsonify({
"error": {
"code": "METHOD_NOT_ALLOWED",
"message": f"The {request.method} method is not allowed for the requested resource.",
"requested_method": request.method,
"allowed_methods": list(sorted_methods),
"path": request.path
}
})
response.status_code = 405
response.headers['Allow'] = ', '.join(sorted_methods)
# 添加缓存控制头
response.headers['Cache-Control'] = 'no-store'
return response
# 路由示例
@app.route('/api/articles', methods=['GET', 'POST'])
def articles_collection():
"""文章集合 - 支持GET和POST"""
if request.method == 'GET':
return jsonify({"articles": []})
else: # POST
return jsonify({"id": 1}), 201
@app.route('/api/articles/', methods=['GET', 'PUT', 'DELETE'])
def article_detail(article_id):
"""文章详情 - 支持GET、PUT、DELETE"""
if request.method == 'GET':
return jsonify({"id": article_id, "title": "Sample"})
elif request.method == 'PUT':
return jsonify({"updated": True}), 200
else: # DELETE
return '', 204
20.3.2 高级方法验证中间件
python
# 方法验证和智能建议中间件
from enum import Enum
from typing import Dict, List, Optional, Set
from dataclasses import dataclass
import re
class HTTPMethod(Enum):
GET = "GET"
POST = "POST"
PUT = "PUT"
PATCH = "PATCH"
DELETE = "DELETE"
HEAD = "HEAD"
OPTIONS = "OPTIONS"
TRACE = "TRACE"
CONNECT = "CONNECT"
@dataclass
class MethodRule:
"""方法规则定义"""
pattern: str
allowed_methods: Set[HTTPMethod]
suggestions: Dict[HTTPMethod, str] = None
reason: str = None
class IntelligentMethodValidator:
"""智能方法验证器"""
def __init__(self):
# 定义RESTful约定
self.restful_rules = [
# 集合端点
MethodRule(
pattern=r'^/api/([^/]+)$',
allowed_methods={HTTPMethod.GET, HTTPMethod.POST, HTTPMethod.OPTIONS},
suggestions={
HTTPMethod.PUT: "Use POST to create resources in a collection",
HTTPMethod.DELETE: "To delete a collection, use a specific resource endpoint"
}
),
# 资源端点
MethodRule(
pattern=r'^/api/([^/]+)/d+$',
allowed_methods={HTTPMethod.GET, HTTPMethod.PUT, HTTPMethod.PATCH, HTTPMethod.DELETE, HTTPMethod.OPTIONS},
suggestions={
HTTPMethod.POST: "Use PUT or PATCH to update an existing resource"
}
),
# 子资源集合
MethodRule(
pattern=r'^/api/([^/]+)/d+/([^/]+)$',
allowed_methods={HTTPMethod.GET, HTTPMethod.POST, HTTPMethod.OPTIONS},
suggestions={
HTTPMethod.PUT: "Use POST to create sub-resources",
HTTPMethod.DELETE: "To delete a sub-resource, use its specific endpoint"
}
),
# 只读端点
MethodRule(
pattern=r'^/api/reports/.+$',
allowed_methods={HTTPMethod.GET, HTTPMethod.HEAD, HTTPMethod.OPTIONS},
reason="This is a read-only reporting endpoint"
)
]
def validate_method(self, path: str, method: HTTPMethod, allowed_methods: Set[HTTPMethod]) -> Dict:
"""验证方法并提供智能反馈"""
# 基本验证
if method in allowed_methods:
return {"valid": True}
# 获取智能建议
suggestions = self.get_suggestions(path, method)
# 检查是否是常见错误
common_errors = self.check_common_errors(path, method)
return {
"valid": False,
"allowed_methods": allowed_methods,
"suggestions": suggestions,
"common_errors": common_errors,
"method_meaning": self.get_method_meaning(method)
}
def get_suggestions(self, path: str, attempted_method: HTTPMethod) -> List[str]:
"""获取针对特定路径和方法的建议"""
suggestions = []
for rule in self.restful_rules:
if re.match(rule.pattern, path):
if rule.suggestions and attempted_method in rule.suggestions:
suggestions.append(rule.suggestions[attempted_method])
if rule.reason:
suggestions.append(f"Reason: {rule.reason}")
# 通用建议
generic_suggestions = {
HTTPMethod.GET: ["Use GET to retrieve resources"],
HTTPMethod.POST: ["Use POST to create new resources"],
HTTPMethod.PUT: ["Use PUT to fully replace existing resources"],
HTTPMethod.PATCH: ["Use PATCH to partially update resources"],
HTTPMethod.DELETE: ["Use DELETE to remove resources"]
}
if attempted_method in generic_suggestions:
suggestions.extend(generic_suggestions[attempted_method])
return suggestions
def check_common_errors(self, path: str, method: HTTPMethod) -> List[Dict]:
"""检查常见错误模式"""
errors = []
# 检查是否是集合与资源混淆
if method == HTTPMethod.PUT and re.match(r'^/api/([^/]+)$', path):
errors.append({
"type": "collection_resource_confusion",
"message": "PUT is typically used for individual resources, not collections",
"suggestion": "Use POST to create a new resource in the collection"
})
# 检查是否误用PATCH
if method == HTTPMethod.PATCH and re.match(r'^/api/([^/]+)$', path):
errors.append({
"type": "patch_on_collection",
"message": "PATCH cannot be applied to collections",
"suggestion": "Apply PATCH to specific resource URLs"
})
# 检查是否使用DELETE删除集合
if method == HTTPMethod.DELETE and re.match(r'^/api/([^/]+)$', path):
errors.append({
"type": "delete_collection",
"message": "Collections typically don't support DELETE",
"suggestion": "Delete individual resources instead"
})
return errors
def get_method_meaning(self, method: HTTPMethod) -> Dict:
"""获取HTTP方法的语义含义"""
meanings = {
HTTPMethod.GET: {
"semantics": "Safe, idempotent retrieval",
"use_case": "Read data without side effects",
"cacheable": True
},
HTTPMethod.POST: {
"semantics": "Create new resource or non-idempotent operation",
"use_case": "Submit data, create resources",
"cacheable": False
},
HTTPMethod.PUT: {
"semantics": "Idempotent replacement",
"use_case": "Update entire resource",
"cacheable": False
},
HTTPMethod.PATCH: {
"semantics": "Partial update",
"use_case": "Update specific fields",
"cacheable": False
},
HTTPMethod.DELETE: {
"semantics": "Remove resource",
"use_case": "Delete existing resource",
"cacheable": False
}
}
return meanings.get(method, {"semantics": "Unknown"})
# Flask中间件集成
validator = IntelligentMethodValidator()
@app.before_request
def validate_http_method():
"""验证HTTP方法"""
if request.method == 'OPTIONS':
return # OPTIONS总是允许的
# 获取当前端点允许的方法
endpoint = request.endpoint
if endpoint:
allowed_methods = method_manager.get_allowed_methods(endpoint)
# 转换为HTTPMethod枚举
allowed_methods_enum = {HTTPMethod(m) for m in allowed_methods}
current_method = HTTPMethod(request.method)
# 验证方法
validation_result = validator.validate_method(
request.path,
current_method,
allowed_methods_enum
)
if not validation_result["valid"]:
# 创建详细的405响应
response = jsonify({
"error": {
"code": "METHOD_NOT_ALLOWED",
"message": f"The {request.method} method is not supported for this endpoint.",
"details": {
"attempted_method": {
"name": request.method,
"semantics": validation_result["method_meaning"]
},
"validation_result": validation_result
}
}
})
response.status_code = 405
response.headers['Allow'] = ', '.join([m.value for m in allowed_methods_enum])
# 添加描述性头部
response.headers['X-Method-Validation'] = 'failed'
if validation_result["suggestions"]:
response.headers['X-Method-Suggestions'] = '; '.join(validation_result["suggestions"])
return response
20.3.3 OPTIONS方法的实现
python
# 完整的OPTIONS方法实现
class OptionsHandler:
"""处理OPTIONS请求，提供API发现功能"""
def __init__(self, app):
self.app = app
self.api_documentation = {}
def generate_options_response(self, endpoint, path):
"""生成OPTIONS响应"""
# 获取允许的方法
allowed_methods = method_manager.get_allowed_methods(endpoint)
# 获取端点文档
endpoint_docs = self.get_endpoint_documentation(endpoint, path)
# 构建响应
response_body = {
"endpoint": endpoint,
"path": path,
"allowed_methods": list(allowed_methods),
"documentation": endpoint_docs,
"links": self.generate_links(path, allowed_methods)
}
# 添加相关端点
related_endpoints = self.find_related_endpoints(path)
if related_endpoints:
response_body["related_endpoints"] = related_endpoints
return response_body
def get_endpoint_documentation(self, endpoint, path):
"""获取端点文档"""
if endpoint in self.api_documentation:
return self.api_documentation[endpoint]
# 动态生成文档
return {
"description": self.infer_endpoint_description(path),
"parameters": self.infer_parameters(path),
"request_format": self.infer_request_format(path),
"response_format": self.infer_response_format(path)
}
def infer_endpoint_description(self, path):
"""推断端点描述"""
if re.match(r'^/api/([^/]+)$', path):
resource = re.match(r'^/api/([^/]+)$', path).group(1)
return f"Collection of {resource} resources"
elif re.match(r'^/api/([^/]+)/d+$', path):
resource = re.match(r'^/api/([^/]+)/d+$', path).group(1)
return f"Individual {resource} resource"
return "API endpoint"
def infer_parameters(self, path):
"""推断参数"""
parameters = {}
# 路径参数
path_params = re.findall(r'<([^>]+)>', path)
for param in path_params:
param_name = param.split(':')[-1] if ':' in param else param
parameters[param_name] = {
"in": "path",
"required": True,
"type": self.infer_param_type(param)
}
return parameters
def infer_param_type(self, param):
"""推断参数类型"""
if param.startswith('int:'):
return "integer"
elif param.startswith('float:'):
return "number"
elif param.startswith('uuid:'):
return "string (UUID)"
else:
return "string"
def infer_request_format(self, path):
"""推断请求格式"""
# 根据路径模式推断
if re.match(r'^/api/([^/]+)$', path):
return {
"POST": {
"content-type": "application/json",
"schema": {
"type": "object",
"properties": {
"id": {"type": "integer", "readOnly": True},
"name": {"type": "string"},
"created_at": {"type": "string", "format": "date-time", "readOnly": True}
}
}
}
}
return {}
def infer_response_format(self, path):
"""推断响应格式"""
return {
"application/json": {
"schema": {"type": "object"}
}
}
def generate_links(self, path, allowed_methods):
"""生成HATEOAS链接"""
links = []
# 自我链接
links.append({
"rel": "self",
"href": path,
"method": "GET" if HTTPMethod.GET in allowed_methods else "OPTIONS"
})
# 父链接
if path != '/':
parent_path = '/'.join(path.split('/')[:-1]) or '/'
links.append({
"rel": "parent",
"href": parent_path,
"method": "GET"
})
# 相关操作链接
if HTTPMethod.POST in allowed_methods:
links.append({
"rel": "create",
"href": path,
"method": "POST"
})
return links
def find_related_endpoints(self, path):
"""查找相关端点"""
related = []
# 查找相同资源的不同操作
if re.match(r'^/api/([^/]+)/d+$', path):
# 这是资源详情，相关的是集合
collection_path = re.match(r'^(/api/[^/]+)/d+$', path).group(1)
related.append({
"relation": "collection",
"path": collection_path,
"methods": ["GET", "POST"]
})
return related
# 集成OPTIONS处理器
options_handler = OptionsHandler(app)
@app.route('/api/', methods=['OPTIONS'])
def handle_options(path):
"""处理所有API端点的OPTIONS请求"""
full_path = f'/api/{path}'
# 查找匹配的端点
adapter = app.url_map.bind('localhost')
try:
endpoint, values = adapter.match(full_path, method='OPTIONS')
except:
# 没有找到匹配的端点
return jsonify({
"error": "Endpoint not found",
"path": full_path
}), 404
# 生成OPTIONS响应
response_data = options_handler.generate_options_response(endpoint, full_path)
# 添加必要的头部
response = jsonify(response_data)
response.headers['Allow'] = ', '.join(response_data['allowed_methods'])
response.headers['Access-Control-Allow-Methods'] = ', '.join(response_data['allowed_methods'])
response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
response.headers['Access-Control-Max-Age'] = '86400' # 24小时
return response
20.4 客户端处理策略
20.4.1 智能方法发现与适配
javascript
// 客户端方法发现和适配系统
class MethodDiscoveryClient {
constructor(baseURL, options = {}) {
this.baseURL = baseURL;
this.options = {
cacheDuration: 5 * 60 * 1000, // 5分钟
autoDiscover: true,
maxRetries: 2,
...options
};
this.methodCache = new Map(); // 缓存端点支持的方法
this.schemaCache = new Map(); // 缓存端点模式
this.fallbackStrategies = new Map(); // 备用策略
}
async discoverMethods(endpoint) {
"""发现端点支持的方法"""
// 检查缓存
const cached = this.getCachedMethods(endpoint);
if (cached) {
return cached;
}
try {
// 发送OPTIONS请求
const response = await this.sendOptionsRequest(endpoint);
if (response.status === 200) {
const data = await response.json();
// 解析允许的方法
const allowedMethods = this.parseAllowedMethods(response, data);
// 缓存结果
this.cacheMethods(endpoint, allowedMethods, data);
return {
success: true,
methods: allowedMethods,
documentation: data
};
}
return {
success: false,
error: `OPTIONS request failed with status ${response.status}`,
methods: this.getDefaultMethods(endpoint)
};
} catch (error) {
console.warn(`Method discovery failed for ${endpoint}:`, error);
return {
success: false,
error: error.message,
methods: this.getDefaultMethods(endpoint)
};
}
}
parseAllowedMethods(response, data) {
"""解析允许的方法"""
const methods = new Set();
// 从Allow头解析
const allowHeader = response.headers.get('Allow');
if (allowHeader) {
allowHeader.split(',').forEach(method => {
methods.add(method.trim());
});
}
// 从响应体解析
if (data && data.allowed_methods) {
data.allowed_methods.forEach(method => methods.add(method));
}
return Array.from(methods);
}
async requestWithDiscovery(endpoint, options = {}) {
"""带有方法发现的请求"""
const { method, ...restOptions } = options;
// 1. 发现端点支持的方法
const discovery = await this.discoverMethods(endpoint);
// 2. 检查请求方法是否允许
if (discovery.methods && !discovery.methods.includes(method)) {
// 方法不允许，尝试找到替代方法
const alternative = this.findAlternativeMethod(method, discovery.methods, endpoint);
if (alternative) {
console.log(`Method ${method} not allowed. Using alternative: ${alternative.method}`);
// 调整请求
const adjustedOptions = this.adjustRequestForAlternative(
method,
alternative.method,
restOptions
);
return this.fetchWithRetry(endpoint, {
method: alternative.method,
...adjustedOptions
});
}
// 没有找到替代方法，抛出错误
throw new MethodNotAllowedError(endpoint, method, discovery.methods);
}
// 3. 执行请求
return this.fetchWithRetry(endpoint, options);
}
findAlternativeMethod(requestedMethod, allowedMethods, endpoint) {
"""查找替代方法"""
// 预定义的替代策略
const alternativeStrategies = {
'PUT': {
alternatives: ['PATCH', 'POST'],
conditions: {
'PATCH': (endpoint) => endpoint.includes('/api/'), // 只对API端点
'POST': (endpoint) => !endpoint.match(//d+$/) // 不对资源ID端点
}
},
'PATCH': {
alternatives: ['PUT'],
conditions: {
'PUT': (endpoint) => true
}
},
'DELETE': {
alternatives: ['POST'],
conditions: {
'POST': (endpoint) => endpoint.includes('/delete') // 特殊删除端点
}
}
};
const strategy = alternativeStrategies[requestedMethod];
if (!strategy) return null;
for (const alternative of strategy.alternatives) {
if (allowedMethods.includes(alternative)) {
const condition = strategy.conditions[alternative];
if (!condition || condition(endpoint)) {
return {
method: alternative,
reason: `Suggested alternative for ${requestedMethod}`
};
}
}
}
return null;
}
adjustRequestForAlternative(originalMethod, alternativeMethod, options) {
"""为替代方法调整请求"""
const adjusted = { ...options };
// 特殊调整
if (originalMethod === 'PUT' && alternativeMethod === 'PATCH') {
// PUT -> PATCH: 可能需要调整请求体
if (adjusted.headers) {
adjusted.headers['X-Original-Method'] = 'PUT';
}
}
if (originalMethod === 'DELETE' && alternativeMethod === 'POST') {
// DELETE -> POST: 添加删除指示器
if (adjusted.body) {
const body = JSON.parse(adjusted.body);
adjusted.body = JSON.stringify({
...body,
_action: 'delete'
});
}
}
return adjusted;
}
async fetchWithRetry(endpoint, options, retryCount = 0) {
"""带重试的fetch"""
try {
const response = await fetch(`${this.baseURL}${endpoint}`, options);
// 处理405响应（即使已经检查过）
if (response.status === 405) {
if (retryCount < this.options.maxRetries) {
// 清除缓存并重试
this.clearCache(endpoint);
return this.requestWithDiscovery(endpoint, options);
}
}
return response;
} catch (error) {
if (retryCount < this.options.maxRetries) {
// 指数退避
const delay = Math.pow(2, retryCount) * 1000;
await this.sleep(delay);
return this.fetchWithRetry(endpoint, options, retryCount + 1);
}
throw error;
}
}
getCachedMethods(endpoint) {
"""获取缓存的方法"""
const cacheEntry = this.methodCache.get(endpoint);
if (cacheEntry && Date.now() - cacheEntry.timestamp < this.options.cacheDuration) {
return cacheEntry.data;
}
return null;
}
cacheMethods(endpoint, methods, documentation) {
"""缓存方法"""
this.methodCache.set(endpoint, {
timestamp: Date.now(),
data: { methods, documentation }
});
}
clearCache(endpoint = null) {
"""清除缓存"""
if (endpoint) {
this.methodCache.delete(endpoint);
} else {
this.methodCache.clear();
}
}
getDefaultMethods(endpoint) {
"""获取默认方法"""
// 基于端点模式推断
if (endpoint.match(//d+$/)) {
return ['GET', 'PUT', 'PATCH', 'DELETE']; // 资源端点
} else {
return ['GET', 'POST']; // 集合端点
}
}
sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}
}
class MethodNotAllowedError extends Error {
constructor(endpoint, method, allowedMethods) {
super(`Method ${method} not allowed for ${endpoint}`);
this.name = 'MethodNotAllowedError';
this.endpoint = endpoint;
this.method = method;
this.allowedMethods = allowedMethods;
}
}
// 使用示例
const client = new MethodDiscoveryClient('https://api.example.com');
// 智能请求
async function updateUser(userId, data) {
try {
const response = await client.requestWithDiscovery(`/api/users/${userId}`, {
method: 'PUT', // 可能自动转为PATCH
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${token}`
},
body: JSON.stringify(data)
});
if (!response.ok) {
throw new Error(`Request failed: ${response.status}`);
}
return await response.json();
} catch (error) {
if (error.name === 'MethodNotAllowedError') {
// 显示用户友好的错误
showMethodErrorUI(error);
return null;
}
throw error;
}
}
// UI组件：显示方法错误
function showMethodErrorUI(error) {
const container = document.getElementById('error-container');
const html = `
❌ Operation Not Supported
The requested operation (${error.method}) is not supported for this resource.
Supported Operations:
${error.allowedMethods.map(method => `
-
${method}
${getMethodDescription(method)}
`).join('')}
`;
container.innerHTML = html;
}
function getMethodDescription(method) {
const descriptions = {
'GET': 'Retrieve data',
'POST': 'Create new resource',
'PUT': 'Replace entire resource',
'PATCH': 'Update partial resource',
'DELETE': 'Remove resource',
'OPTIONS': 'Get supported methods'
};
return descriptions[method] || '';
}
20.5 监控与分析
20.5.1 405错误分析系统
python
# 405错误监控与分析
from datetime import datetime, timedelta
from collections import defaultdict, Counter
import json
from typing import Dict, List, Any
from dataclasses import dataclass, asdict
import re
@dataclass
class MethodNotAllowedEvent:
timestamp: datetime
endpoint: str
attempted_method: str
allowed_methods: List[str]
client_ip: str
user_agent: str
referer: str
api_version: str = None
user_id: str = None
request_body_preview: str = None
def to_dict(self):
data = asdict(self)
data['timestamp'] = self.timestamp.isoformat()
return data
class MethodNotAllowedAnalyzer:
def __init__(self):
self.events = []
self.patterns = defaultdict(lambda: {
'total': 0,
'by_method': Counter(),
'by_client': Counter(),
'by_api_version': Counter(),
'first_seen': None,
'last_seen': None
})
# 常见错误模式检测器
self.error_patterns = {
'collection_put': re.compile(r'^/api/([^/]+)$'),
'resource_post': re.compile(r'^/api/([^/]+)/d+$'),
'incorrect_http_method': re.compile(r'PUT|POST|PATCH|DELETE', re.I),
'legacy_api': re.compile(r'/v1/|/legacy/'),
}
def record_event(self, event: MethodNotAllowedEvent):
"""记录405事件"""
self.events.append(event)
# 更新模式统计
pattern = self.categorize_endpoint(event.endpoint)
pattern_data = self.patterns[pattern]
pattern_data['total'] += 1
pattern_data['by_method'][event.attempted_method] += 1
pattern_data['by_client'][event.client_ip] += 1
if event.api_version:
pattern_data['by_api_version'][event.api_version] += 1
pattern_data['last_seen'] = event.timestamp
if not pattern_data['first_seen']:
pattern_data['first_seen'] = event.timestamp
# 实时分析
self.realtime_analysis(event)
# 持久化
self.persist_event(event)
# 检查是否需要警报
self.check_for_alerts(event, pattern_data)
def categorize_endpoint(self, endpoint: str) -> str:
"""对端点进行分类"""
# 提取模式：将ID等替换为占位符
pattern = re.sub(r'/d+', '/{id}', endpoint)
pattern = re.sub(r'/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', '/{uuid}', pattern, flags=re.I)
pattern = re.sub(r'/[a-f0-9]{32}', '/{hash}', pattern)
return pattern
def realtime_analysis(self, event: MethodNotAllowedEvent):
"""实时分析405事件"""
insights = []
# 1. 检测常见RESTful错误
if self.error_patterns['collection_put'].match(event.endpoint) and event.attempted_method == 'PUT':
insights.append({
'type': 'collection_put_error',
'message': 'PUT method used on collection endpoint',
'suggestion': 'Use POST for creating resources in collections'
})
if self.error_patterns['resource_post'].match(event.endpoint) and event.attempted_method == 'POST':
insights.append({
'type': 'resource_post_error',
'message': 'POST method used on resource endpoint',
'suggestion': 'Use PUT or PATCH for updating existing resources'
})
# 2. 检测API版本问题
if self.error_patterns['legacy_api'].search(event.endpoint):
insights.append({
'type': 'legacy_api_access',
'message': 'Access to legacy API version',
'suggestion': 'Migrate to newer API version'
})
# 3. 检测可能的恶意扫描
if self.is_scanning_attempt(event):
insights.append({
'type': 'possible_scanning',
'message': 'Possible scanning attempt detected',
'severity': 'high'
})
# 记录洞察
if insights:
self.record_insights(event, insights)
def is_scanning_attempt(self, event: MethodNotAllowedEvent) -> bool:
"""检测是否是扫描尝试"""
# 检查短时间内多种方法尝试
recent_events = self.get_recent_events(event.client_ip, minutes=5)
if len(recent_events) > 10:
# 短时间内大量405错误
return True
# 检查是否尝试了危险方法
dangerous_methods = ['CONNECT', 'TRACE']
if event.attempted_method in dangerous_methods:
return True
# 检查是否尝试了多种方法
unique_methods = set(e.attempted_method for e in recent_events)
if len(unique_methods) > 5:
return True
return False
def get_recent_events(self, client_ip: str, minutes: int = 5) -> List[MethodNotAllowedEvent]:
"""获取客户端最近的405事件"""
cutoff = datetime.utcnow() - timedelta(minutes=minutes)
return [
e for e in self.events
if e.client_ip == client_ip and e.timestamp > cutoff
]
def check_for_alerts(self, event: MethodNotAllowedEvent, pattern_data: Dict):
"""检查是否需要发送警报"""
# 1. 高频405模式警报
if pattern_data['total'] > 100:
self.send_alert({
'type': 'high_frequency_405',
'pattern': event.endpoint,
'count': pattern_data['total'],
'timestamp': datetime.utcnow().isoformat(),
'severity': 'medium'
})
# 2. 客户端异常行为警报
if pattern_data['by_client'][event.client_ip] > 20:
self.send_alert({
'type': 'client_405_abuse',
'client_ip': event.client_ip,
'count': pattern_data['by_client'][event.client_ip],
'timestamp': datetime.utcnow().isoformat(),
'severity': 'high'
})
# 3. API版本弃用警报
if event.api_version and 'v1' in event.api_version:
self.send_alert({
'type': 'legacy_api_usage',
'endpoint': event.endpoint,
'api_version': event.api_version,
'method': event.attempted_method,
'timestamp': datetime.utcnow().isoformat(),
'severity': 'low'
})
def generate_report(self, period: str = '24h') -> Dict[str, Any]:
"""生成405分析报告"""
if period == '24h':
cutoff = datetime.utcnow() - timedelta(hours=24)
elif period == '7d':
cutoff = datetime.utcnow() - timedelta(days=7)
elif period == '30d':
cutoff = datetime.utcnow() - timedelta(days=30)
else:
cutoff = datetime.utcnow() - timedelta(hours=24)
period_events = [e for e in self.events if e.timestamp > cutoff]
report = {
'period': period,
'time_range': {
'start': cutoff.isoformat(),
'end': datetime.utcnow().isoformat()
},
'summary': {
'total_405s': len(period_events),
'unique_endpoints': len(set(e.endpoint for e in period_events)),
'unique_clients': len(set(e.client_ip for e in period_events)),
'methods_attempted': list(set(e.attempted_method for e in period_events))
},
'top_patterns': sorted(
[(pattern, data['total']) for pattern, data in self.patterns.items()],
key=lambda x: x[1],
reverse=True
)[:10],
'top_methods': Counter(e.attempted_method for e in period_events).most_common(),
'top_clients': Counter(e.client_ip for e in period_events).most_common(10),
'hourly_distribution': self.get_hourly_distribution(period_events),
'insights': self.generate_insights(period_events)
}
return report
def get_hourly_distribution(self, events: List[MethodNotAllowedEvent]) -> Dict[str, int]:
"""获取按小时分布"""
distribution = defaultdict(int)
for event in events:
hour = event.timestamp.strftime('%H:00')
distribution[hour] += 1
return dict(sorted(distribution.items()))
def generate_insights(self, events: List[MethodNotAllowedEvent]) -> List[Dict]:
"""生成洞察和建议"""
insights = []
if not events:
return insights
# 按端点分组
endpoint_groups = defaultdict(list)
for event in events:
endpoint_groups[event.endpoint].append(event)
# 分析每个频繁出现405的端点
for endpoint, endpoint_events in endpoint_groups.items():
if len(endpoint_events) > 5: # 阈值
insight = self.analyze_endpoint_pattern(endpoint, endpoint_events)
if insight:
insights.append(insight)
# 总体洞察
overall_insights = self.analyze_overall_patterns(events)
insights.extend(overall_insights)
return insights[:20] # 限制数量
def analyze_endpoint_pattern(self, endpoint: str, events: List[MethodNotAllowedEvent]) -> Dict:
"""分析特定端点的405模式"""
methods_attempted = Counter(e.attempted_method for e in events)
clients = Counter(e.client_ip for e in events)
insight = {
'endpoint': endpoint,
'total_attempts': len(events),
'top_methods': methods_attempted.most_common(3),
'unique_clients': len(clients),
'suggestions': []
}
# 根据模式生成建议
if re.match(r'^/api/([^/]+)$', endpoint):
# 集合端点
if 'PUT' in methods_attempted:
insight['suggestions'].append(
'Consider adding PUT support or clarifying documentation'
)
elif re.match(r'^/api/([^/]+)/d+$', endpoint):
# 资源端点
if 'POST' in methods_attempted:
insight['suggestions'].append(
'Clients are trying to POST to resource endpoints. '
'This might indicate confusion about RESTful conventions.'
)
# 如果有大量不同的客户端出现相同错误
if len(clients) > 10:
insight['suggestions'].append(
'Multiple clients are making the same error. '
'Consider improving API documentation or client SDKs.'
)
return insight if insight['suggestions'] else None
def persist_event(self, event: MethodNotAllowedEvent):
"""持久化事件到文件"""
with open('405_method_errors.jsonl', 'a') as f:
f.write(json.dumps(event.to_dict()) + '
')
def record_insights(self, event: MethodNotAllowedEvent, insights: List[Dict]):
"""记录洞察"""
insight_record = {
'timestamp': datetime.utcnow().isoformat(),
'event': event.to_dict(),
'insights': insights
}
with open('405_insights.jsonl', 'a') as f:
f.write(json.dumps(insight_record) + '
')
def send_alert(self, alert: Dict):
"""发送警报"""
# 这里可以集成各种通知系统
print(f"[405 Alert] {json.dumps(alert)}")
# 记录到文件
with open('405_alerts.jsonl', 'a') as f:
f.write(json.dumps(alert) + '
')
20.5.2 405错误监控仪表板
javascript
// React组件：405监控仪表板
import React, { useState, useEffect } from 'react';
import {
BarChart, Bar, LineChart, Line, PieChart, Pie, Cell,
XAxis, YAxis, CartesianGrid, Tooltip, Legend, ResponsiveContainer
} from 'recharts';
function MethodNotAllowedDashboard() {
const [timeRange, setTimeRange] = useState('24h');
const [stats, setStats] = useState(null);
const [realtimeEvents, setRealtimeEvents] = useState([]);
const [topPatterns, setTopPatterns] = useState([]);
// 获取统计数据
useEffect(() => {
const fetchStats = async () => {
const response = await fetch(`/api/monitor/405-stats?range=${timeRange}`);
const data = await response.json();
setStats(data);
// 处理模式数据用于图表
if (data.top_patterns) {
setTopPatterns(data.top_patterns.map(([pattern, count]) => ({
pattern,
count,
shortPattern: pattern.length > 30 ? pattern.substring(0, 30) + '...' : pattern
})));
}
};
fetchStats();
const interval = setInterval(fetchStats, 60000); // 每分钟更新
return () => clearInterval(interval);
}, [timeRange]);
// WebSocket连接接收实时事件
useEffect(() => {
const ws = new WebSocket('wss://api.example.com/monitor/405-events');
ws.onmessage = (event) => {
const eventData = JSON.parse(event.data);
setRealtimeEvents(prev => [
{ ...eventData, id: Date.now() },
...prev.slice(0, 49) // 保留最近50个
]);
};
return () => ws.close();
}, []);
if (!stats) {
return 加载中...;
}
return (
405 Method Not Allowed 监控
{['1h', '24h', '7d', '30d'].map(range => (
))}
{/* 概览卡片 */}
总405错误数
{stats.summary.total_405s.toLocaleString()}
过去24小时
受影响端点
{stats.summary.unique_endpoints}
{topPatterns.length} 个主要模式
影响的客户端
{stats.summary.unique_clients}
{stats.top_clients?.[0]?.[1] || 0} 次来自最活跃客户端
错误方法
{stats.summary.methods_attempted.length}
最常见: {stats.top_methods?.[0]?.[0] || 'N/A'}
{/* 图表区域 */}
405错误趋势
({ hour, count })) : []}>
错误方法分布
({ method, count })) || []}
cx="50%"
cy="50%"
labelLine={false}
label={({ method, percent }) => `${method}: ${(percent * 100).toFixed(1)}%`}
outerRadius={80}
fill="#8884d8"
dataKey="count"
>
{stats.top_methods?.map((entry, index) => (
|
))}
最常见端点模式
[
value,
props.payload.pattern.length > 30 ?
`${props.payload.pattern.substring(0, 30)}...` :
props.payload.pattern
]}
/>
客户端分布
({ ip, count })) || []}>
{/* 实时事件流 */}
实时405事件
时间
端点
尝试的方法
客户端IP
允许的方法
{realtimeEvents.map(event => (
{new Date(event.timestamp).toLocaleTimeString()}
{event.endpoint.length > 40 ?
event.endpoint.substring(0, 40) + '...' :
event.endpoint}
{event.attempted_method}
{event.client_ip}
{event.user_agent && (
{event.user_agent}
)}
{event.allowed_methods?.map(method => (
{method}
))}
))}
{/* 洞察和建议 */}
{stats.insights && stats.insights.length > 0 && (
洞察和建议
{stats.insights.map((insight, index) => (
端点模式: {insight.endpoint}
总错误数: {insight.total_attempts}
影响客户端: {insight.unique_clients}
常见错误方法:
{insight.top_methods?.map(([method, count]) => (
{method} ({count})
))}
{insight.suggestions && insight.suggestions.length > 0 && (
建议:
{insight.suggestions.map((suggestion, i) => (
- {suggestion}
))}
)}
))}
)}
{/* 导出和操作 */}
);
}
20.6 安全考虑
20.6.1 防止方法枚举攻击
python
# 防止HTTP方法枚举攻击
class MethodEnumerationProtection:
def __init__(self, config=None):
self.config = config or {
'max_attempts_per_ip': 50, # 每个IP最大尝试次数
'time_window_minutes': 5, # 时间窗口
'ban_duration_minutes': 30, # 封禁时长
'sensitive_endpoints': [ # 敏感端点
'/admin',
'/api/internal',
'/debug',
'/config'
]
}
self.attempt_tracker = defaultdict(list)
self.banned_ips = {}
self.suspicious_patterns = [
r'.(php|asp|jsp|aspx)$', # 脚本文件
r'/(bin|etc|usr|var|opt)/', # 系统目录
r'.(bak|old|backup|tar|gz)$', # 备份文件
]
def check_request(self, request):
"""检查请求，防止方法枚举攻击"""
client_ip = request.remote_addr
path = request.path
method = request.method
# 1. 检查是否被封禁
if self.is_banned(client_ip):
return {
'allowed': False,
'reason': 'IP temporarily banned',
'status': 429
}
# 2. 检查是否是敏感端点
if self.is_sensitive_endpoint(path):
# 对敏感端点进行更严格的检查
if not self.validate_sensitive_access(client_ip, path, method):
return {
'allowed': False,
'reason': 'Sensitive endpoint access denied',
'status': 403
}
# 3. 检查方法枚举尝试
if self.is_method_enumeration(client_ip, path, method):
self.record_suspicious_activity(client_ip, 'method_enumeration', {
'path': path,
'method': method,
'timestamp': datetime.utcnow().isoformat()
})
# 触发保护
self.ban_ip(client_ip)
return {
'allowed': False,
'reason': 'Method enumeration detected',
'status': 429
}
# 4. 检查可疑路径模式
if self.is_suspicious_path(path):
self.record_suspicious_activity(client_ip, 'suspicious_path', {
'path': path,
'method': method,
'timestamp': datetime.utcnow().isoformat()
})
# 返回简化的405响应，不泄露信息
return {
'allowed': False,
'reason': 'Suspicious request',
'status': 404, # 故意返回404而不是405
'simplified_response': True
}
return {'allowed': True}
def is_banned(self, client_ip):
"""检查IP是否被封禁"""
if client_ip in self.banned_ips:
ban_time = self.banned_ips[client_ip]
ban_duration = timedelta(minutes=self.config['ban_duration_minutes'])
if datetime.utcnow() - ban_time < ban_duration:
return True
else:
# 封禁过期
del self.banned_ips[client_ip]
return False
def ban_ip(self, client_ip):
"""封禁IP"""
self.banned_ips[client_ip] = datetime.utcnow()
# 记录安全事件
self.log_security_event('ip_banned', {
'client_ip': client_ip,
'reason': 'Method enumeration',
'timestamp': datetime.utcnow().isoformat(),
'duration_minutes': self.config['ban_duration_minutes']
})
def is_sensitive_endpoint(self, path):
"""检查是否是敏感端点"""
path_lower = path.lower()
return any(ep in path_lower for ep in self.config['sensitive_endpoints'])
def validate_sensitive_access(self, client_ip, path, method):
"""验证对敏感端点的访问"""
# 这里可以实现更复杂的逻辑，如：
# - 检查白名单
# - 验证API密钥
# - 检查地理位置
# 简单示例：只允许GET和OPTIONS方法
allowed_methods = {'GET', 'OPTIONS'}
return method in allowed_methods
def is_method_enumeration(self, client_ip, path, method):
"""检查是否是方法枚举尝试"""
# 记录尝试
key = f"{client_ip}:{path}"
self.attempt_tracker[key].append({
'method': method,
'timestamp': datetime.utcnow()
})
# 清理旧记录
cutoff = datetime.utcnow() - timedelta(minutes=self.config['time_window_minutes'])
self.attempt_tracker[key] = [
attempt for attempt in self.attempt_tracker[key]
if attempt['timestamp'] > cutoff
]
# 检查尝试次数
if len(self.attempt_tracker[key]) > self.config['max_attempts_per_ip']:
return True
# 检查是否尝试了多种方法
unique_methods = set(attempt['method'] for attempt in self.attempt_tracker[key])
if len(unique_methods) > 5: # 尝试了5种以上不同方法
return True
# 检查是否尝试了危险方法
dangerous_methods = {'CONNECT', 'TRACE'}
if any(attempt['method'] in dangerous_methods for attempt in self.attempt_tracker[key]):
return True
return False
def is_suspicious_path(self, path):
"""检查是否是可疑路径"""
path_lower = path.lower()
# 检查可疑模式
for pattern in self.suspicious_patterns:
if re.search(pattern, path_lower):
return True
# 检查路径遍历
if '..' in path or '%2e%2e' in path_lower:
return True
# 检查过长的路径
if len(path) > 500:
return True
return False
def secure_405_response(self, path, method, allowed_methods, request, simplified=False):
"""生成安全的405响应"""
client_ip = request.remote_addr
if simplified:
# 简化响应，不泄露信息
response = jsonify({
'error': {
'code': 'NOT_FOUND',
'message': 'Resource not found'
}
})
response.status_code = 404
response.headers['Allow'] = 'GET, POST' # 通用列表
else:
# 正常405响应
response = jsonify({
'error': {
'code': 'METHOD_NOT_ALLOWED',
'message': f'Method {method} not allowed'
}
})
response.status_code = 405
response.headers['Allow'] = ', '.join(allowed_methods)
# 添加安全头部
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
# 记录访问（安全版本）
self.log_405_access(client_ip, path, method, simplified)
return response
def log_405_access(self, client_ip, path, method, simplified):
"""记录405访问（安全版本）"""
# 脱敏处理
safe_path = self.sanitize_path(path)
log_entry = {
'timestamp': datetime.utcnow().isoformat(),
'client_ip': client_ip[:8] + '...', # 部分隐藏IP
'path': safe_path,
'method': method,
'simplified_response': simplified
}
with open('secure_405_logs.jsonl', 'a') as f:
f.write(json.dumps(log_entry) + '
')
def sanitize_path(self, path):
"""脱敏路径，移除敏感信息"""
# 移除查询参数
clean_path = path.split('?')[0]
# 替换ID等敏感信息
clean_path = re.sub(r'/d+', '/{id}', clean_path)
clean_path = re.sub(r'/[a-f0-9]{32}', '/{hash}', clean_path, flags=re.I)
return clean_path
def record_suspicious_activity(self, client_ip, activity_type, data):
"""记录可疑活动"""
event = {
'type': activity_type,
'client_ip': client_ip,
'data': data,
'timestamp': datetime.utcnow().isoformat()
}
with open('security_events.jsonl', 'a') as f:
f.write(json.dumps(event) + '
')
# 发送警报
self.send_security_alert(event)
def log_security_event(self, event_type, data):
"""记录安全事件"""
event = {
'type': event_type,
'data': data,
'timestamp': datetime.utcnow().isoformat()
}
with open('security_logs.jsonl', 'a') as f:
f.write(json.dumps(event) + '
')
def send_security_alert(self, event):
"""发送安全警报"""
# 这里可以集成各种通知系统
print(f"[Security Alert] {json.dumps(event)}")
# 示例：发送到Slack
if event['type'] in ['method_enumeration', 'ip_banned']:
self.send_slack_alert(event)
# 集成保护机制
protection = MethodEnumerationProtection()
@app.before_request
def apply_method_protection():
"""应用方法保护"""
if request.method not in ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS']:
# 不常见的方法，立即检查
result = protection.check_request(request)
if not result['allowed']:
if result.get('simplified_response'):
return protection.secure_405_response(
request.path, request.method, [], request, simplified=True
)
else:
return jsonify({'error': result['reason']}), result.get('status', 403)
@app.errorhandler(405)
def handle_405_with_protection(e):
"""带保护的405处理"""
# 先进行安全检查
result = protection.check_request(request)
if not result['allowed'] and result.get('simplified_response'):
return protection.secure_405_response(
request.path, request.method, [], request, simplified=True
)
# 正常405处理
allowed_methods = method_manager.get_allowed_methods(request.endpoint)
return protection.secure_405_response(
request.path, request.method, allowed_methods, request, simplified=False
)
20.7 性能优化
20.7.1 高效的方法验证
python
# 高效的方法验证和缓存系统
from functools import lru_cache
from typing import Set, Tuple
import time
import threading
class OptimizedMethodValidator:
"""优化的方法验证器，带缓存和并发支持"""
def __init__(self, app, cache_size=1000, ttl=300):
self.app = app
self.cache_size = cache_size
self.ttl = ttl # 缓存生存时间（秒）
# 线程安全的缓存
self.cache = {}
self.cache_timestamps = {}
self.cache_lock = threading.Lock()
# 路由前缀树，用于快速匹配
self.route_trie = self.build_route_trie()
# 热点路由统计
self.route_stats = defaultdict(int)
self.hot_routes = set()
def build_route_trie(self):
"""构建路由前缀树"""
trie = {}
for rule in self.app.url_map.iter_rules():
path = rule.rule
methods = frozenset(rule.methods)
# 构建trie
current = trie
for segment in path.split('/'):
if segment.startswith('<'):
# 参数段，使用通配符
if '*' not in current:
current['*'] = {}
current = current['*']
else:
if segment not in current:
current[segment] = {}
current = current[segment]
# 存储端点的方法
current['__methods__'] = methods
current['__endpoint__'] = rule.endpoint
return trie
def find_route_methods(self, path: str) -> Tuple[str, Set[str]]:
"""使用trie查找路由和方法"""
segments = path.strip('/').split('/')
current = self.route_trie
params = {}
for segment in segments:
if segment in current:
current = current[segment]
elif '*' in current:
current = current['*']
params[segment] = segment
else:
return None, set()
if '__methods__' in current:
return current.get('__endpoint__'), current['__methods__']
return None, set()
@lru_cache(maxsize=1000)
def get_allowed_methods_cached(self, endpoint: str) -> Set[str]:
"""带缓存的方法获取"""
# 实际查询逻辑
for rule in self.app.url_map.iter_rules():
if rule.endpoint == endpoint:
return frozenset(rule.methods)
return frozenset()
def validate_method_optimized(self, path: str, method: str) -> dict:
"""优化的方法验证"""
start_time = time.time()
# 检查缓存
cache_key = (path, method)
with self.cache_lock:
if cache_key in self.cache:
cache_entry = self.cache[cache_key]
if time.time() - self.cache_timestamps[cache_key] < self.ttl:
# 缓存命中
self.cache_timestamps[cache_key] = time.time()
return cache_entry
# 缓存未命中，使用trie查找
endpoint, allowed_methods = self.find_route_methods(path)
if not endpoint:
# 没有找到路由
result = {
'valid': False,
'allowed_methods': {'OPTIONS'},
'reason': 'ROUTE_NOT_FOUND',
'endpoint': None
}
else:
# 找到路由，检查方法
is_valid = method in allowed_methods
# 更新路由统计
self.route_stats[endpoint] += 1
if self.route_stats[endpoint] > 100:
self.hot_routes.add(endpoint)
result = {
'valid': is_valid,
'allowed_methods': allowed_methods,
'reason': 'METHOD_NOT_ALLOWED' if not is_valid else 'OK',
'endpoint': endpoint
}
# 缓存结果（如果值得缓存）
if self.should_cache_result(result, path):
with self.cache_lock:
# 清理过期缓存
self.cleanup_cache()
# 存储新缓存
self.cache[cache_key] = result
self.cache_timestamps[cache_key] = time.time()
# 记录性能数据
elapsed = time.time() - start_time
if elapsed > 0.01: # 超过10ms
self.log_slow_validation(path, method, elapsed)
return result
def should_cache_result(self, result: dict, path: str) -> bool:
"""判断是否应该缓存结果"""
# 不缓存的情况：
# 1. 路由未找到
if not result['endpoint']:
return False
# 2. 路径包含参数（动态路径）
if re.search(r'/d+|/[a-f0-9]{32}|/[^/]+/[^/]+/[^/]+', path):
return False
# 3. 热点路由总是缓存
if result['endpoint'] in self.hot_routes:
return True
# 4. 静态路径或简单API路径
if re.match(r'^/api/([^/]+)$', path) or re.match(r'^/static/', path):
return True
return False
def cleanup_cache(self):
"""清理过期缓存"""
if len(self.cache) < self.cache_size * 1.5:
return # 缓存未满，不需要清理
current_time = time.time()
expired_keys = []
for key, timestamp in self.cache_timestamps.items():
if current_time - timestamp > self.ttl:
expired_keys.append(key)
# 移除过期缓存
for key in expired_keys:
self.cache.pop(key, None)
self.cache_timestamps.pop(key, None)
# 如果仍然太大，移除最旧的
if len(self.cache) > self.cache_size:
sorted_keys = sorted(
self.cache_timestamps.items(),
key=lambda x: x[1]
)[:len(self.cache) - self.cache_size]
for key, _ in sorted_keys:
self.cache.pop(key, None)
self.cache_timestamps.pop(key, None)
def log_slow_validation(self, path: str, method: str, elapsed: float):
"""记录慢速验证"""
log_entry = {
'timestamp': datetime.utcnow().isoformat(),
'path': path,
'method': method,
'elapsed_ms': round(elapsed * 1000, 2),
'type': 'slow_method_validation'
}
with open('performance_logs.jsonl', 'a') as f:
f.write(json.dumps(log_entry) + '
')
# 如果特别慢，发送警报
if elapsed > 0.1: # 超过100ms
self.send_performance_alert(log_entry)
def get_performance_report(self) -> dict:
"""获取性能报告"""
with self.cache_lock:
cache_stats = {
'total_cached': len(self.cache),
'cache_hits': getattr(self, 'cache_hits', 0),
'cache_misses': getattr(self, 'cache_misses', 0),
'hot_routes': len(self.hot_routes)
}
# 计算命中率
total = cache_stats['cache_hits'] + cache_stats['cache_misses']
cache_stats['hit_rate'] = cache_stats['cache_hits'] / total if total > 0 else 0
return {
'cache': cache_stats,
'route_stats': dict(sorted(
self.route_stats.items(),
key=lambda x: x[1],
reverse=True
)[:10])
}
def send_performance_alert(self, log_entry: dict):
"""发送性能警报"""
alert = {
'type': 'slow_method_validation',
'data': log_entry,
'timestamp': datetime.utcnow().isoformat(),
'severity': 'warning'
}
print(f"[Performance Alert] {json.dumps(alert)}")
# 集成优化的验证器
optimized_validator = OptimizedMethodValidator(app)
@app.before_request
def fast_method_validation():
"""快速方法验证中间件"""
# 跳过OPTIONS和HEAD方法
if request.method in ['OPTIONS', 'HEAD']:
return
# 使用优化验证器
validation_result = optimized_validator.validate_method_optimized(
request.path,
request.method
)
if not validation_result['valid']:
# 立即返回405，避免进入视图函数
response = jsonify({
'error': {
'code': 'METHOD_NOT_ALLOWED',
'message': f'Method {request.method} is not allowed for this endpoint',
'allowed_methods': list(validation_result['allowed_methods'])
}
})
response.status_code = 405
response.headers['Allow'] = ', '.join(validation_result['allowed_methods'])
return response
# 性能监控端点
@app.route('/api/performance/method-validation')
def get_method_validation_stats():
"""获取方法验证性能统计"""
report = optimized_validator.get_performance_report()
return jsonify(report)
20.8 特殊场景处理
20.8.1 API版本迁移的方法处理
python
# API版本迁移中的方法处理
class APIVersionMigrationHandler:
"""处理API版本迁移中的方法变化"""
def __init__(self):
self.version_mappings = {
'v1': {
'deprecated_methods': {
'PUT': {
'since': '2024-01-01',
'alternative': 'PATCH',
'sunset': '2025-01-01'
},
'POST': {
'since': '2024-03-01',
'alternative': 'PUT',
'contexts': ['/api/v1/users//permissions']
}
},
'new_methods': {
'PATCH': {
'introduced': '2024-01-01',
'description': 'Partial updates'
}
}
},
'v2': {
'deprecated_methods': {},
'new_methods': {
'PATCH': {},
'HEAD': {
'introduced': '2024-01-01',
'description': 'Resource metadata'
}
}
}
}
self.version_redirects = {
'/api/v1/users': '/api/v2/users',
'/api/v1/products': '/api/v2/products'
}
def handle_method_for_version(self, path: str, method: str, request) -> dict:
"""处理特定版本的方法"""
# 提取API版本
version = self.extract_api_version(path)
if not version:
return {'proceed': True}
# 检查是否有重定向
redirected_path = self.check_version_redirect(path)
if redirected_path:
return {
'proceed': False,
'action': 'redirect',
'new_path': redirected_path,
'status': 301
}
# 检查方法弃用
deprecation_info = self.check_method_deprecation(version, path, method)
if deprecation_info:
return {
'proceed': True,
'deprecation': deprecation_info
}
# 检查方法支持
if not self.is_method_supported(version, path, method):
# 提供版本特定的405响应
return {
'proceed': False,
'action': 'version_specific_405',
'version': version,
'method': method
}
return {'proceed': True}
def extract_api_version(self, path: str) -> str:
"""提取API版本"""
match = re.match(r'^/api/(v[0-9]+)/', path)
return match.group(1) if match else None
def check_version_redirect(self, path: str) -> str:
"""检查版本重定向"""
for old_path, new_path in self.version_redirects.items():
if path.startswith(old_path):
# 替换路径前缀
return path.replace(old_path, new_path, 1)
return None
def check_method_deprecation(self, version: str, path: str, method: str) -> dict:
"""检查方法是否已弃用"""
if version not in self.version_mappings:
return None
deprecated_methods = self.version_mappings[version].get('deprecated_methods', {})
if method in deprecated_methods:
info = deprecated_methods[method]
# 检查是否适用于当前上下文
if 'contexts' in info:
if not any(path.startswith(ctx.replace('', '')) for ctx in info['contexts']):
return None
return {
'method': method,
'deprecated_since': info['since'],
'alternative': info.get('alternative'),
'sunset': info.get('sunset'),
'message': f'Method {method} is deprecated in {version}'
}
return None
def is_method_supported(self, version: str, path: str, method: str) -> bool:
"""检查方法在特定版本中是否支持"""
# 获取基础允许的方法
endpoint = request.endpoint
if endpoint:
allowed_methods = method_manager.get_allowed_methods(endpoint)
return method in allowed_methods
# 版本特定的检查
if version == 'v1':
# v1不支持PATCH
if method == 'PATCH':
return False
return True
def create_version_specific_405(self, version: str, method: str, path: str) -> tuple:
"""创建版本特定的405响应"""
# 获取允许的方法
endpoint = request.endpoint
allowed_methods = method_manager.get_allowed_methods(endpoint) if endpoint else set()
# 版本特定的调整
if version == 'v1':
# v1中移除PATCH
allowed_methods = {m for m in allowed_methods if m != 'PATCH'}
response_data = {
"error": {
"code": "METHOD_NOT_ALLOWED",
"message": f"Method {method} is not supported in API {version}",
"api_version": version,
"allowed_methods": list(allowed_methods),
"version_specific": True
}
}
# 添加版本迁移信息
if version == 'v1':
response_data["error"]["migration_note"] = (
"Consider migrating to API v2 for additional features and methods."
)
response_data["error"]["v2_endpoint"] = f"/api/v2{path[7:]}" # 移除/v1前缀
response = jsonify(response_data)
response.status_code = 405
response.headers['Allow'] = ', '.join(allowed_methods)
response.headers['X-API-Version'] = version
response.headers['Deprecation'] = 'true'
return response
def create_deprecation_response(self, deprecation_info: dict) -> tuple:
"""创建弃用警告响应"""
response = jsonify({
"warning": {
"code": "METHOD_DEPRECATED",
"message": deprecation_info['message'],
"deprecated_method": deprecation_info['method'],
"deprecated_since": deprecation_info['deprecated_since'],
"alternative_method": deprecation_info.get('alternative'),
"sunset_date": deprecation_info.get('sunset'),
"documentation": f"https://api.example.com/docs/migration-guide"
}
})
response.status_code = 200 # 仍然处理请求，但添加警告
response.headers['Deprecation'] = f'date="{deprecation_info["deprecated_since"]}"'
if deprecation_info.get('sunset'):
response.headers['Sunset'] = deprecation_info['sunset']
if deprecation_info.get('alternative'):
response.headers['Link'] = (
f'; rel="successor-version"; type="application/json"'
)
return response
# 集成版本迁移处理器
version_handler = APIVersionMigrationHandler()
@app.before_request
def handle_api_version_migration():
"""处理API版本迁移"""
# 检查API版本和方法
result = version_handler.handle_method_for_version(
request.path, request.method, request
)
if not result['proceed']:
if result['action'] == 'redirect':
# 重定向到新版本
response = jsonify({
"error": {
"code": "VERSION_REDIRECT",
"message": f"API endpoint has moved to {result['new_path']}",
"new_location": result['new_path']
}
})
response.status_code = result['status']
response.headers['Location'] = result['new_path']
return response
elif result['action'] == 'version_specific_405':
# 版本特定的405
return version_handler.create_version_specific_405(
result['version'], result['method'], request.path
)
elif 'deprecation' in result:
# 方法已弃用，添加警告头部
request.deprecation_info = result['deprecation']
@app.after_request
def add_deprecation_headers(response):
"""添加弃用头部"""
if hasattr(request, 'deprecation_info'):
deprecation_info = request.deprecation_info
response.headers['Deprecation'] = f'date="{deprecation_info["deprecated_since"]}"'
if deprecation_info.get('sunset'):
response.headers['Sunset'] = deprecation_info['sunset']
# 如果响应是JSON，可以在响应体中添加警告
if response.headers.get('Content-Type', '').startswith('application/json'):
try:
data = json.loads(response.get_data(as_text=True))
if isinstance(data, dict):
data['_warning'] = {
'deprecated_method': deprecation_info['method'],
'message': deprecation_info['message'],
'alternative': deprecation_info.get('alternative')
}
response.set_data(json.dumps(data))
except:
pass # 忽略JSON解析错误
return response
第21章：408 Request Timeout - 请求超时管理深度分析
21.1 定义与语义
408 Request Timeout 状态码表示服务器在等待请求的完整传输时超过了预设的时间限制。该状态码的核心含义是：
-
服务器没有在期望的时间内收到完整的请求
-
连接可能仍然保持打开状态
-
客户端可以在稍后时间用相同的请求重试
关键特性：
-
不是客户端错误：而是客户端与服务器之间的通信问题
-
服务器主动关闭连接：由于超时而终止等待
-
与504 Gateway Timeout的区别：504是上游服务器超时，408是客户端请求超时
协议要求：
-
服务器应该发送一个关闭连接的响应
-
可以包含描述超时的响应体
-
客户端可以重试相同的请求
21.2 触发场景分类
21.2.1 网络层超时场景
python
# 网络层超时分析
class NetworkTimeoutAnalyzer:
def __init__(self):
self.timeout_scenarios = {
# 1. 连接建立超时
'tcp_handshake': {
'description': 'TCP三次握手未在时间内完成',
'typical_timeout': 60, # 秒
'common_causes': [
'网络拥塞',
'防火墙阻止',
'DNS解析失败'
]
},
# 2. TLS握手超时
'tls_handshake': {
'description': 'TLS握手过程超时',
'typical_timeout': 30, # 秒
'common_causes': [
'证书链验证复杂',
'客户端/服务器CPU负载高',
'弱加密套件协商'
]
},
# 3. 请求头传输超时
'headers_transmission': {
'description': '请求头传输时间过长',
'typical_timeout': 30, # 秒
'common_causes': [
'头部过大（如Cookie）',
'网络带宽限制',
'代理服务器处理延迟'
]
},
# 4. 请求体传输超时
'body_transmission': {
'description': '请求体传输时间过长',
'typical_timeout': 60, # 秒
'common_causes': [
'大文件上传',
'慢速网络连接',
'客户端处理延迟'
]
},
# 5. 慢速loris攻击
'slow_loris': {
'description': '客户端缓慢发送数据，占用连接',
'typical_timeout': 300, # 秒
'common_causes': [
'恶意攻击',
'客户端实现缺陷',
'网络问题'
]
}
}
def analyze_timeout(self, request_data, network_stats):
"""分析超时原因"""
timeout_type = self.determine_timeout_type(request_data, network_stats)
if timeout_type:
scenario = self.timeout_scenarios[timeout_type]
return {
'type': timeout_type,
'description': scenario['description'],
'causes': self.rank_causes(request_data, scenario['common_causes']),
'recommendations': self.generate_recommendations(timeout_type)
}
return None
def determine_timeout_type(self, request_data, network_stats):
"""确定超时类型"""
# 分析连接时间
if network_stats.get('connection_time', 0) > 60:
return 'tcp_handshake'
# 分析TLS握手时间
if network_stats.get('tls_handshake_time', 0) > 30:
return 'tls_handshake'
# 分析请求头大小
headers_size = sum(len(k) + len(v) for k, v in request_data.get('headers', {}).items())
if headers_size > 8192: # 8KB
return 'headers_transmission'
# 分析请求体大小和传输速率
body_size = request_data.get('content_length', 0)
if body_size > 10 * 1024 * 1024: # 10MB
if network_stats.get('transfer_rate', 0) < 1024: # < 1KB/s
return 'body_transmission'
# 检查慢速loris模式
if self.is_slow_loris_pattern(request_data, network_stats):
return 'slow_loris'
return None
def is_slow_loris_pattern(self, request_data, network_stats):
"""检测慢速loris攻击模式"""
# 检查请求是否分段传输
if network_stats.get('chunk_count', 0) > 10:
# 检查传输间隔
intervals = network_stats.get('chunk_intervals', [])
if intervals:
avg_interval = sum(intervals) / len(intervals)
if 1 < avg_interval < 10: # 每秒发送少量数据
return True
# 检查Keep-Alive滥用
if network_stats.get('keep_alive_requests', 0) > 100:
return True
return False
def rank_causes(self, request_data, possible_causes):
"""对可能原因进行排序"""
ranked = []
# 基于请求特征排序
for cause in possible_causes:
score = 0
if cause == '网络拥塞' and request_data.get('high_latency', False):
score += 3
if cause == '大文件上传' and request_data.get('content_length', 0) > 5 * 1024 * 1024:
score += 3
if cause == '恶意攻击' and self.is_slow_loris_pattern(request_data, {}):
score += 5
ranked.append((cause, score))
# 按分数排序
ranked.sort(key=lambda x: x[1], reverse=True)
return [cause for cause, score in ranked if score > 0]
def generate_recommendations(self, timeout_type):
"""生成推荐解决方案"""
recommendations = {
'tcp_handshake': [
'检查网络连接稳定性',
'验证防火墙配置',
'使用连接池减少握手次数'
],
'tls_handshake': [
'优化证书链',
'使用更快的加密算法',
'启用TLS会话恢复'
],
'headers_transmission': [
'减少Cookie大小',
'压缩请求头',
'使用HTTP/2头部压缩'
],
'body_transmission': [
'分块上传大文件',
'启用压缩',
'增加超时时间'
],
'slow_loris': [
'配置请求速率限制',
'限制最小传输速率',
'使用Web应用防火墙'
]
}
return recommendations.get(timeout_type, [])
21.2.2 应用层超时场景
python
# 应用层超时场景分析
class ApplicationTimeoutAnalyzer:
def __init__(self):
self.request_stages = {
'parsing': {
'description': '请求解析阶段',
'timeout': 5, # 秒
'metrics': ['request_size', 'complexity']
},
'authentication': {
'description': '身份验证阶段',
'timeout': 10,
'metrics': ['auth_method', 'token_validation']
},
'authorization': {
'description': '授权检查阶段',
'timeout': 5,
'metrics': ['permission_checks', 'role_validation']
},
'validation': {
'description': '请求验证阶段',
'timeout': 10,
'metrics': ['data_size', 'validation_rules']
},
'processing': {
'description': '业务处理阶段',
'timeout': 30,
'metrics': ['complexity', 'external_calls']
},
'response': {
'description': '响应生成阶段',
'timeout': 10,
'metrics': ['response_size', 'serialization']
}
}
def track_request_stage(self, request_id, stage, start_time):
"""跟踪请求阶段"""
return {
'request_id': request_id,
'stage': stage,
'start_time': start_time,
'timeout': self.request_stages[stage]['timeout']
}
def check_stage_timeout(self, stage_info):
"""检查阶段是否超时"""
current_time = time.time()
elapsed = current_time - stage_info['start_time']
if elapsed > stage_info['timeout']:
return {
'timed_out': True,
'stage': stage_info['stage'],
'elapsed': elapsed,
'timeout': stage_info['timeout'],
'description': self.request_stages[stage_info['stage']]['description']
}
return {'timed_out': False}
def analyze_staleness(self, request_data):
"""分析请求陈旧性"""
# 检查请求是否在队列中等待太久
received_time = request_data.get('received_time')
start_time = request_data.get('start_time')
if received_time and start_time:
queue_time = start_time - received_time
if queue_time > 30: # 在队列中等待超过30秒
return {
'stale': True,
'queue_time': queue_time,
'recommendation': '增加处理能力或减少队列长度'
}
return {'stale': False}
def generate_timeout_response(self, timeout_info):
"""生成超时响应"""
response = {
'error': {
'code': 'REQUEST_TIMEOUT',
'message': 'Request processing timed out',
'details': {
'stage': timeout_info['stage'],
'elapsed_seconds': round(timeout_info['elapsed'], 2),
'timeout_seconds': timeout_info['timeout'],
'description': timeout_info['description']
}
}
}
# 添加恢复建议
if timeout_info['stage'] == 'processing':
response['error']['suggestions'] = [
'简化请求复杂性',
'使用异步处理',
'增加超时时间配置'
]
return response
21.3 详细实现与最佳实践
21.3.1 完整的超时管理中间件
python
# 完整的超时管理中间件实现
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor, TimeoutError
from contextlib import contextmanager
from typing import Dict, Optional, Callable, Any
import signal
import threading
class TimeoutManager:
"""全面的超时管理器"""
def __init__(self, config: Dict = None):
self.config = {
'connection_timeout': 60, # 连接建立超时
'read_timeout': 60, # 读取请求超时
'write_timeout': 60, # 写入响应超时
'request_timeout': 120, # 总请求处理超时
'keep_alive_timeout': 15, # Keep-Alive超时
'max_header_size': 8192, # 最大头部大小
'max_body_size': 10 * 1024 * 1024, # 最大请求体大小
'slow_request_threshold': 10, # 慢请求阈值（秒）
** (config or {})
}
self.active_connections = {}
self.slow_requests = []
self.timeout_stats = {
'total_timeouts': 0,
'by_stage': {},
'by_client': {}
}
# 初始化超时处理器
self.init_timeout_handlers()
def init_timeout_handlers(self):
"""初始化超时处理器"""
self.handlers = {
'connection': self.handle_connection_timeout,
'read': self.handle_read_timeout,
'write': self.handle_write_timeout,
'request': self.handle_request_timeout,
'keep_alive': self.handle_keep_alive_timeout
}
@contextmanager
def request_context(self, client_ip: str, request_id: str):
"""请求上下文管理器"""
start_time = time.time()
connection_info = {
'client_ip': client_ip,
'request_id': request_id,
'start_time': start_time,
'stage': 'initial',
'last_activity': start_time,
'bytes_received': 0,
'bytes_sent': 0
}
self.active_connections[request_id] = connection_info
try:
yield connection_info
finally:
# 清理连接信息
if request_id in self.active_connections:
elapsed = time.time() - start_time
if elapsed > self.config['slow_request_threshold']:
self.record_slow_request(connection_info, elapsed)
del self.active_connections[request_id]
def monitor_connection(self, request_id: str):
"""监控连接状态"""
def monitor():
while request_id in self.active_connections:
conn = self.active_connections[request_id]
idle_time = time.time() - conn['last_activity']
# 检查读超时
if idle_time > self.config['read_timeout'] and conn['stage'] == 'reading':
self.handle_timeout('read', conn)
break
# 检查总请求超时
total_time = time.time() - conn['start_time']
if total_time > self.config['request_timeout']:
self.handle_timeout('request', conn)
break
time.sleep(1) # 每秒检查一次
# 启动监控线程
thread = threading.Thread(target=monitor, daemon=True)
thread.start()
return thread
def handle_timeout(self, timeout_type: str, connection_info: Dict):
"""处理超时"""
handler = self.handlers.get(timeout_type)
if handler:
handler(connection_info)
# 记录统计
self.record_timeout(timeout_type, connection_info)
# 发送警报
self.send_timeout_alert(timeout_type, connection_info)
def handle_connection_timeout(self, connection_info: Dict):
"""处理连接超时"""
# 记录连接超时详情
log_data = {
'type': 'connection_timeout',
'client_ip': connection_info['client_ip'],
'request_id': connection_info['request_id'],
'stage': connection_info['stage'],
'duration': time.time() - connection_info['start_time'],
'timestamp': time.time()
}
self.log_timeout_event(log_data)
# 关闭连接
self.close_connection(connection_info)
def handle_read_timeout(self, connection_info: Dict):
"""处理读超时"""
log_data = {
'type': 'read_timeout',
'client_ip': connection_info['client_ip'],
'request_id': connection_info['request_id'],
'stage': connection_info['stage'],
'bytes_received': connection_info['bytes_received'],
'idle_time': time.time() - connection_info['last_activity'],
'timestamp': time.time()
}
self.log_timeout_event(log_data)
# 检查是否是慢速loris攻击
if self.is_slow_loris_attack(connection_info):
self.handle_slow_loris(connection_info)
def handle_write_timeout(self, connection_info: Dict):
"""处理写超时"""
log_data = {
'type': 'write_timeout',
'client_ip': connection_info['client_ip'],
'request_id': connection_info['request_id'],
'stage': connection_info['stage'],
'bytes_sent': connection_info['bytes_sent'],
'timestamp': time.time()
}
self.log_timeout_event(log_data)
def handle_request_timeout(self, connection_info: Dict):
"""处理请求处理超时"""
log_data = {
'type': 'request_timeout',
'client_ip': connection_info['client_ip'],
'request_id': connection_info['request_id'],
'stage': connection_info['stage'],
'total_duration': time.time() - connection_info['start_time'],
'timestamp': time.time()
}
self.log_timeout_event(log_data)
# 生成408响应
self.send_408_response(connection_info)
def handle_keep_alive_timeout(self, connection_info: Dict):
"""处理Keep-Alive超时"""
log_data = {
'type': 'keep_alive_timeout',
'client_ip': connection_info['client_ip'],
'request_id': connection_info['request_id'],
'idle_time': time.time() - connection_info['last_activity'],
'timestamp': time.time()
}
self.log_timeout_event(log_data)
def is_slow_loris_attack(self, connection_info: Dict) -> bool:
"""检测慢速loris攻击"""
# 检查接收速率
if connection_info['bytes_received'] > 0:
duration = time.time() - connection_info['start_time']
rate = connection_info['bytes_received'] / duration
if rate < 100: # 小于100字节/秒
return True
# 检查长时间保持连接但没有数据
idle_time = time.time() - connection_info['last_activity']
if idle_time > 30 and connection_info['bytes_received'] < 100:
return True
return False
def handle_slow_loris(self, connection_info: Dict):
"""处理慢速loris攻击"""
# 添加到黑名单
self.add_to_blacklist(connection_info['client_ip'])
# 立即关闭连接
self.close_connection_immediately(connection_info)
# 记录安全事件
self.log_security_event('slow_loris_attack', connection_info)
def send_408_response(self, connection_info: Dict):
"""发送408响应"""
response = self.create_408_response(connection_info)
# 在实际应用中，这里会通过连接发送响应
# 为了示例，我们只记录
self.log_response(response)
def create_408_response(self, connection_info: Dict) -> Dict:
"""创建408响应"""
response = {
'status_code': 408,
'headers': {
'Content-Type': 'application/json',
'Connection': 'close',
'X-Request-Timeout': 'true',
'X-Timeout-Stage': connection_info['stage'],
'X-Request-ID': connection_info['request_id']
},
'body': {
'error': {
'code': 'REQUEST_TIMEOUT',
'message': 'Server timeout waiting for the request',
'request_id': connection_info['request_id'],
'client_ip': connection_info['client_ip'],
'timeout_type': 'request_processing',
'duration_seconds': round(time.time() - connection_info['start_time'], 2),
'recommendations': [
'Simplify the request',
'Retry with exponential backoff',
'Check network connectivity'
]
}
}
}
return response
def record_timeout(self, timeout_type: str, connection_info: Dict):
"""记录超时统计"""
self.timeout_stats['total_timeouts'] += 1
# 按阶段统计
stage = connection_info['stage']
if stage not in self.timeout_stats['by_stage']:
self.timeout_stats['by_stage'][stage] = 0
self.timeout_stats['by_stage'][stage] += 1
# 按客户端统计
client_ip = connection_info['client_ip']
if client_ip not in self.timeout_stats['by_client']:
self.timeout_stats['by_client'][client_ip] = 0
self.timeout_stats['by_client'][client_ip] += 1
def record_slow_request(self, connection_info: Dict, duration: float):
"""记录慢请求"""
slow_request = {
'request_id': connection_info['request_id'],
'client_ip': connection_info['client_ip'],
'duration': duration,
'stage': connection_info['stage'],
'timestamp': time.time()
}
self.slow_requests.append(slow_request)
# 保持最近1000个慢请求
if len(self.slow_requests) > 1000:
self.slow_requests = self.slow_requests[-1000:]
def log_timeout_event(self, log_data: Dict):
"""记录超时事件"""
# 在实际应用中，这里会记录到日志系统
print(f"[Timeout Event] {log_data}")
# 写入文件日志
with open('timeout_events.log', 'a') as f:
f.write(json.dumps(log_data) + '
')
def log_response(self, response: Dict):
"""记录响应"""
with open('408_responses.log', 'a') as f:
f.write(json.dumps(response) + '
')
def add_to_blacklist(self, client_ip: str):
"""添加到黑名单"""
# 在实际应用中，这里会更新黑名单存储
with open('blacklist.log', 'a') as f:
f.write(f"{client_ip},{time.time()}
")
def log_security_event(self, event_type: str, data: Dict):
"""记录安全事件"""
event = {
'type': event_type,
'data': data,
'timestamp': time.time()
}
with open('security_events.log', 'a') as f:
f.write(json.dumps(event) + '
')
def close_connection(self, connection_info: Dict):
"""关闭连接"""
# 在实际应用中，这里会关闭网络连接
print(f"Closing connection for request {connection_info['request_id']}")
def close_connection_immediately(self, connection_info: Dict):
"""立即关闭连接"""
# 强制关闭连接
print(f"Immediately closing connection for suspected attack: {connection_info['request_id']}")
def get_statistics(self) -> Dict:
"""获取统计信息"""
return {
'active_connections': len(self.active_connections),
'total_timeouts': self.timeout_stats['total_timeouts'],
'timeouts_by_stage': self.timeout_stats['by_stage'],
'slow_requests_count': len(self.slow_requests),
'recent_slow_requests': self.slow_requests[-10:] if self.slow_requests else []
}
21.3.2 异步超时处理框架
python
# 异步超时处理框架
import asyncio
from typing import Optional, Callable, Any, Dict
import aiohttp
from aiohttp import web
import async_timeout
class AsyncTimeoutFramework:
"""异步超时处理框架"""
def __init__(self, app: web.Application):
self.app = app
self.timeout_config = {
'global_timeout': 120,
'stage_timeouts': {
'request_parsing': 5,
'authentication': 10,
'validation': 5,
'processing': 30,
'response_generation': 10
},
'retry_policy': {
'max_retries': 3,
'backoff_factor': 2,
'max_backoff': 60
}
}
# 中间件设置
self.setup_middleware()
def setup_middleware(self):
"""设置超时中间件"""
@web.middleware
async def timeout_middleware(request: web.Request, handler: Callable):
"""超时中间件"""
request_start = asyncio.get_event_loop().time()
request_id = request.headers.get('X-Request-ID', str(uuid.uuid4()))
# 创建超时上下文
timeout_context = {
'request_id': request_id,
'start_time': request_start,
'current_stage': 'initial',
'timeout_tasks': []
}
request['timeout_context'] = timeout_context
try:
# 设置全局超时
async with async_timeout.timeout(self.timeout_config['global_timeout']):
# 执行处理
response = await self.execute_with_stage_timeouts(request, handler)
return response
except asyncio.TimeoutError:
# 全局超时
elapsed = asyncio.get_event_loop().time() - request_start
return await self.handle_global_timeout(request, elapsed)
except Exception as e:
# 其他异常
return await self.handle_timeout_exception(request, e)
finally:
# 清理超时任务
await self.cleanup_timeout_tasks(timeout_context)
self.app.middlewares.append(timeout_middleware)
async def execute_with_stage_timeouts(self, request: web.Request, handler: Callable):
"""分阶段执行请求处理"""
timeout_context = request['timeout_context']
# 阶段1: 请求解析
timeout_context['current_stage'] = 'request_parsing'
await self.start_stage_timeout('request_parsing', timeout_context)
# 解析请求
try:
data = await request.json()
request['parsed_data'] = data
except:
pass # 非JSON请求
# 阶段2: 认证
timeout_context['current_stage'] = 'authentication'
await self.start_stage_timeout('authentication', timeout_context)
# 执行认证
await self.authenticate_request(request)
# 阶段3: 验证
timeout_context['current_stage'] = 'validation'
await self.start_stage_timeout('validation', timeout_context)
# 验证请求
await self.validate_request(request)
# 阶段4: 处理
timeout_context['current_stage'] = 'processing'
await self.start_stage_timeout('processing', timeout_context)
# 执行业务逻辑
response = await handler(request)
# 阶段5: 响应生成
timeout_context['current_stage'] = 'response_generation'
await self.start_stage_timeout('response_generation', timeout_context)
# 添加超时相关头部
response.headers.update({
'X-Request-Timeout-Context': json.dumps({
'request_id': timeout_context['request_id'],
'total_time': round(asyncio.get_event_loop().time() - timeout_context['start_time'], 3)
})
})
return response
async def start_stage_timeout(self, stage: str, timeout_context: Dict):
"""启动阶段超时监控"""
timeout_seconds = self.timeout_config['stage_timeouts'].get(stage, 10)
task = asyncio.create_task(
self.monitor_stage_timeout(stage, timeout_seconds, timeout_context)
)
timeout_context['timeout_tasks'].append(task)
async def monitor_stage_timeout(self, stage: str, timeout_seconds: float, timeout_context: Dict):
"""监控阶段超时"""
try:
await asyncio.sleep(timeout_seconds)
# 如果超时后仍然在这个阶段，触发超时处理
if timeout_context['current_stage'] == stage:
await self.handle_stage_timeout(stage, timeout_context)
except asyncio.CancelledError:
# 任务被取消，正常情况
pass
async def handle_stage_timeout(self, stage: str, timeout_context: Dict):
"""处理阶段超时"""
# 记录超时事件
await self.log_stage_timeout(stage, timeout_context)
# 根据阶段决定处理方式
if stage == 'processing':
# 处理阶段超时，可能需要中断处理
await self.interrupt_processing(timeout_context)
async def handle_global_timeout(self, request: web.Request, elapsed: float):
"""处理全局超时"""
timeout_context = request.get('timeout_context', {})
current_stage = timeout_context.get('current_stage', 'unknown')
# 创建408响应
response_data = {
'error': {
'code': 'REQUEST_TIMEOUT',
'message': 'Request processing exceeded maximum time limit',
'details': {
'elapsed_seconds': round(elapsed, 2),
'max_timeout_seconds': self.timeout_config['global_timeout'],
'stage_at_timeout': current_stage,
'request_id': timeout_context.get('request_id')
},
'suggestions': [
'Simplify the request payload',
'Use pagination for large datasets',
'Implement asynchronous processing'
]
}
}
response = web.json_response(
response_data,
status=408,
headers={
'Connection': 'close',
'X-Request-Timeout': 'true',
'X-Timeout-Stage': current_stage
}
)
# 记录超时
await self.log_global_timeout(request, elapsed, current_stage)
return response
async def handle_timeout_exception(self, request: web.Request, exception: Exception):
"""处理超时异常"""
# 记录异常
await self.log_timeout_exception(request, exception)
# 返回500错误
return web.json_response(
{'error': 'Internal server error'},
status=500
)
async def cleanup_timeout_tasks(self, timeout_context: Dict):
"""清理超时任务"""
for task in timeout_context.get('timeout_tasks', []):
if not task.done():
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
async def authenticate_request(self, request: web.Request):
"""认证请求"""
# 实际认证逻辑
pass
async def validate_request(self, request: web.Request):
"""验证请求"""
# 实际验证逻辑
pass
async def interrupt_processing(self, timeout_context: Dict):
"""中断处理"""
# 发送中断信号或设置标志
timeout_context['interrupted'] = True
async def log_stage_timeout(self, stage: str, timeout_context: Dict):
"""记录阶段超时"""
log_entry = {
'type': 'stage_timeout',
'stage': stage,
'request_id': timeout_context.get('request_id'),
'timestamp': time.time(),
'timeout_seconds': self.timeout_config['stage_timeouts'].get(stage, 10)
}
# 在实际应用中，这里会记录到日志系统
print(f"[Stage Timeout] {log_entry}")
async def log_global_timeout(self, request: web.Request, elapsed: float, stage: str):
"""记录全局超时"""
log_entry = {
'type': 'global_timeout',
'request_id': request.headers.get('X-Request-ID'),
'client_ip': request.remote,
'path': request.path,
'method': request.method,
'elapsed_seconds': round(elapsed, 2),
'stage_at_timeout': stage,
'timestamp': time.time()
}
# 记录到文件
with open('global_timeouts.log', 'a') as f:
f.write(json.dumps(log_entry) + '
')
async def log_timeout_exception(self, request: web.Request, exception: Exception):
"""记录超时异常"""
log_entry = {
'type': 'timeout_exception',
'request_id': request.headers.get('X-Request-ID'),
'exception_type': type(exception).__name__,
'exception_message': str(exception),
'timestamp': time.time()
}
with open('timeout_exceptions.log', 'a') as f:
f.write(json.dumps(log_entry) + '
')
21.3.3 智能重试与回退策略
python
# 智能重试与回退策略
class SmartRetryStrategy:
"""智能重试策略管理器"""
def __init__(self, config: Dict = None):
self.config = {
'max_retries': 3,
'base_delay': 1.0, # 基础延迟（秒）
'max_delay': 60.0, # 最大延迟
'jitter': 0.1, # 抖动因子
'backoff_factor': 2.0, # 退避因子
'timeout_multiplier': 1.5, # 超时乘数
'retryable_errors': [408, 429, 500, 502, 503, 504],
'circuit_breaker': {
'failure_threshold': 5,
'reset_timeout': 60,
'half_open_max_requests': 3
},
** (config or {})
}
self.circuit_breakers = {}
self.retry_stats = defaultdict(lambda: {
'total_attempts': 0,
'successful_retries': 0,
'failed_retries': 0,
'total_delay': 0.0
})
async def execute_with_retry(self, request_func: Callable, request_data: Dict,
context: Dict = None) -> Any:
"""执行带重试的请求"""
context = context or {}
retry_count = 0
last_exception = None
while retry_count <= self.config['max_retries']:
attempt_start = time.time()
# 检查熔断器
circuit_key = self.get_circuit_key(request_data)
if not self.allow_request(circuit_key):
raise CircuitBreakerOpenError(f"Circuit breaker open for {circuit_key}")
try:
# 计算本次尝试的超时时间
timeout = self.calculate_timeout(retry_count)
# 执行请求
async with async_timeout.timeout(timeout):
result = await request_func(request_data)
# 请求成功
if retry_count > 0:
self.record_successful_retry(circuit_key, retry_count)
self.reset_circuit_breaker(circuit_key)
return result
except asyncio.TimeoutError:
# 超时异常
last_exception = TimeoutError(f"Request timeout after {timeout} seconds")
await self.handle_timeout(circuit_key, retry_count)
except Exception as e:
# 其他异常
last_exception = e
if not self.should_retry(e):
raise e
await self.handle_error(circuit_key, e, retry_count)
# 准备重试
retry_count += 1
if retry_count <= self.config['max_retries']:
# 计算延迟
delay = self.calculate_delay(retry_count)
self.retry_stats[circuit_key]['total_delay'] += delay
# 记录重试
await self.log_retry_attempt(request_data, retry_count, delay, last_exception)
# 等待延迟
await asyncio.sleep(delay)
# 所有重试都失败
raise MaxRetriesExceededError(
f"Max retries ({self.config['max_retries']}) exceeded",
last_exception=last_exception
)
def get_circuit_key(self, request_data: Dict) -> str:
"""获取熔断器键"""
# 基于端点和方法创建键
endpoint = request_data.get('endpoint', 'unknown')
method = request_data.get('method', 'GET')
return f"{method}:{endpoint}"
def allow_request(self, circuit_key: str) -> bool:
"""检查是否允许请求"""
if circuit_key not in self.circuit_breakers:
return True
cb = self.circuit_breakers[circuit_key]
if cb['state'] == 'open':
# 检查是否应该进入半开状态
if time.time() - cb['opened_at'] >= self.config['circuit_breaker']['reset_timeout']:
cb['state'] = 'half_open'
cb['half_open_attempts'] = 0
return True
return False
elif cb['state'] == 'half_open':
# 限制半开状态的并发请求
if cb['half_open_attempts'] >= self.config['circuit_breaker']['half_open_max_requests']:
return False
cb['half_open_attempts'] += 1
return True
return True # closed状态
def calculate_timeout(self, retry_count: int) -> float:
"""计算超时时间"""
# 指数退避：每次重试增加超时时间
base_timeout = 30.0 # 基础超时30秒
timeout = base_timeout * (self.config['timeout_multiplier'] ** retry_count)
return min(timeout, 300.0) # 最大5分钟
def calculate_delay(self, retry_count: int) -> float:
"""计算重试延迟"""
# 指数退避算法
delay = self.config['base_delay'] * (self.config['backoff_factor'] ** (retry_count - 1))
# 添加随机抖动
jitter = random.uniform(1 - self.config['jitter'], 1 + self.config['jitter'])
delay *= jitter
# 限制最大延迟
return min(delay, self.config['max_delay'])
def should_retry(self, exception: Exception) -> bool:
"""检查是否应该重试"""
# 检查HTTP状态码
if hasattr(exception, 'status_code'):
return exception.status_code in self.config['retryable_errors']
# 检查异常类型
retryable_exceptions = [
TimeoutError,
ConnectionError,
asyncio.TimeoutError
]
for exc_type in retryable_exceptions:
if isinstance(exception, exc_type):
return True
return False
async def handle_timeout(self, circuit_key: str, retry_count: int):
"""处理超时"""
# 记录超时
self.record_failure(circuit_key)
# 检查是否应该打开熔断器
if self.should_open_circuit(circuit_key):
self.open_circuit_breaker(circuit_key)
async def handle_error(self, circuit_key: str, error: Exception, retry_count: int):
"""处理错误"""
# 记录错误
self.record_failure(circuit_key)
# 检查是否应该打开熔断器
if self.should_open_circuit(circuit_key):
self.open_circuit_breaker(circuit_key)
def record_failure(self, circuit_key: str):
"""记录失败"""
if circuit_key not in self.circuit_breakers:
self.circuit_breakers[circuit_key] = {
'state': 'closed',
'failure_count': 0,
'success_count': 0,
'last_failure': None,
'opened_at': None
}
cb = self.circuit_breakers[circuit_key]
cb['failure_count'] += 1
cb['last_failure'] = time.time()
# 检查失败阈值
if (cb['state'] == 'closed' and
cb['failure_count'] >= self.config['circuit_breaker']['failure_threshold']):
self.open_circuit_breaker(circuit_key)
def record_successful_retry(self, circuit_key: str, retry_count: int):
"""记录成功重试"""
self.retry_stats[circuit_key]['successful_retries'] += 1
self.retry_stats[circuit_key]['total_attempts'] += retry_count + 1
# 更新熔断器
if circuit_key in self.circuit_breakers:
cb = self.circuit_breakers[circuit_key]
cb['success_count'] += 1
# 如果在半开状态成功，关闭熔断器
if cb['state'] == 'half_open' and cb['success_count'] >= 3:
self.close_circuit_breaker(circuit_key)
def should_open_circuit(self, circuit_key: str) -> bool:
"""检查是否应该打开熔断器"""
if circuit_key not in self.circuit_breakers:
return False
cb = self.circuit_breakers[circuit_key]
# 检查失败率
total_requests = cb['failure_count'] + cb['success_count']
if total_requests < 10: # 需要最小样本量
return False
failure_rate = cb['failure_count'] / total_requests
return failure_rate > 0.5 # 失败率超过50%
def open_circuit_breaker(self, circuit_key: str):
"""打开熔断器"""
cb = self.circuit_breakers[circuit_key]
cb['state'] = 'open'
cb['opened_at'] = time.time()
# 记录熔断事件
self.log_circuit_event(circuit_key, 'opened')
def close_circuit_breaker(self, circuit_key: str):
"""关闭熔断器"""
cb = self.circuit_breakers[circuit_key]
cb['state'] = 'closed'
cb['failure_count'] = 0
cb['success_count'] = 0
# 记录熔断事件
self.log_circuit_event(circuit_key, 'closed')
def reset_circuit_breaker(self, circuit_key: str):
"""重置熔断器"""
if circuit_key in self.circuit_breakers:
cb = self.circuit_breakers[circuit_key]
if cb['state'] == 'closed':
# 逐渐减少失败计数（衰减）
cb['failure_count'] = max(0, cb['failure_count'] - 1)
async def log_retry_attempt(self, request_data: Dict, retry_count: int,
delay: float, exception: Exception):
"""记录重试尝试"""
log_entry = {
'timestamp': time.time(),
'request_id': request_data.get('request_id'),
'endpoint': request_data.get('endpoint'),
'method': request_data.get('method'),
'retry_count': retry_count,
'delay_seconds': round(delay, 2),
'exception_type': type(exception).__name__,
'exception_message': str(exception)[:200]
}
# 在实际应用中，这里会记录到日志系统
print(f"[Retry Attempt] {log_entry}")
def log_circuit_event(self, circuit_key: str, event: str):
"""记录熔断事件"""
log_entry = {
'timestamp': time.time(),
'circuit_key': circuit_key,
'event': event,
'state': self.circuit_breakers[circuit_key]['state']
}
with open('circuit_breaker_events.log', 'a') as f:
f.write(json.dumps(log_entry) + '
')
def get_retry_statistics(self) -> Dict:
"""获取重试统计"""
stats = {
'circuit_breakers': {},
'retry_stats': dict(self.retry_stats),
'summary': {
'total_circuits': len(self.circuit_breakers),
'open_circuits': sum(1 for cb in self.circuit_breakers.values()
if cb['state'] == 'open'),
'half_open_circuits': sum(1 for cb in self.circuit_breakers.values()
if cb['state'] == 'half_open')
}
}
for key, cb in self.circuit_breakers.items():
stats['circuit_breakers'][key] = {
'state': cb['state'],
'failure_count': cb['failure_count'],
'success_count': cb['success_count']
}
return stats
class CircuitBreakerOpenError(Exception):
"""熔断器打开异常"""
pass
class MaxRetriesExceededError(Exception):
"""最大重试次数超出异常"""
def __init__(self, message, last_exception=None):
super().__init__(message)
self.last_exception = last_exception
21.4 客户端处理策略
21.4.1 自适应重试与降级机制
javascript
// 客户端自适应重试策略
class AdaptiveRetryClient {
constructor(config = {}) {
this.config = {
maxRetries: 3,
baseDelay: 1000,
maxDelay: 30000,
timeout: 30000,
timeoutMultiplier: 1.5,
backoffFactor: 2,
jitter: 0.2,
circuitBreaker: {
failureThreshold: 5,
resetTimeout: 60000,
halfOpenMaxRequests: 2
},
fallbackStrategies: {
timeout: 'return_cached',
network_error: 'retry_with_exponential_backoff',
server_error: 'use_alternative_endpoint'
},
...config
};
this.circuitBreakers = new Map();
this.requestHistory = new Map();
this.fallbackCache = new Map();
this.metrics = {
totalRequests: 0,
successfulRequests: 0,
failedRequests: 0,
timeouts: 0,
retries: 0
};
// 网络质量检测
this.networkQuality = {
latency: 0,
throughput: 0,
reliability: 1.0
};
// 初始化网络监控
this.initNetworkMonitoring();
}
async fetchWithRetry(url, options = {}) {
const requestId = this.generateRequestId();
const startTime = Date.now();
this.metrics.totalRequests++;
// 创建请求上下文
const context = {
requestId,
url,
method: options.method || 'GET',
startTime,
attempt: 0,
delays: [],
errors: []
};
// 检查熔断器
const circuitKey = this.getCircuitKey(url, options.method);
if (!this.allowRequest(circuitKey)) {
return await this.executeFallback('circuit_breaker_open', context, options);
}
// 执行带重试的请求
try {
const result = await this.executeWithAdaptiveRetry(context, options);
// 请求成功
this.metrics.successfulRequests++;
this.recordSuccess(circuitKey);
// 更新网络质量指标
this.updateNetworkQuality(context, true);
return result;
} catch (error) {
// 请求失败
this.metrics.failedRequests++;
if (error.name === 'TimeoutError') {
this.metrics.timeouts++;
}
// 记录失败
this.recordFailure(circuitKey, error);
// 更新网络质量指标
this.updateNetworkQuality(context, false);
// 执行降级策略
return await this.executeFallback(error.name, context, options, error);
}
}
async executeWithAdaptiveRetry(context, options) {
let lastError;
for (let attempt = 0; attempt <= this.config.maxRetries; attempt++) {
context.attempt = attempt;
try {
// 计算本次尝试的超时时间
const timeout = this.calculateTimeout(attempt);
// 创建带超时的fetch请求
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), timeout);
const fetchOptions = {
...options,
signal: controller.signal
};
// 执行请求
const startAttempt = Date.now();
const response = await fetch(context.url, fetchOptions);
const attemptDuration = Date.now() - startAttempt;
clearTimeout(timeoutId);
// 检查HTTP状态码
if (!response.ok) {
const error = new Error(`HTTP ${response.status}`);
error.statusCode = response.status;
// 决定是否重试
if (this.shouldRetry(error, attempt)) {
lastError = error;
context.errors.push(error);
await this.handleRetry(context, attempt, error);
continue;
}
throw error;
}
// 请求成功
if (attempt > 0) {
this.metrics.retries++;
}
return response;
} catch (error) {
lastError = error;
context.errors.push(error);
// 检查是否应该重试
if (attempt < this.config.maxRetries && this.shouldRetry(error, attempt)) {
await this.handleRetry(context, attempt, error);
continue;
}
// 不再重试，抛出错误
break;
}
}
throw lastError || new Error('Request failed');
}
calculateTimeout(attempt) {
// 自适应超时：基于历史响应时间
const baseTimeout = this.config.timeout;
// 考虑网络质量
const networkFactor = Math.max(0.5, Math.min(2.0,
this.networkQuality.latency / 100)); // 假设基准延迟100ms
// 指数退避
const backoffTimeout = baseTimeout *
Math.pow(this.config.timeoutMultiplier, attempt) *
networkFactor;
return Math.min(backoffTimeout, this.config.maxDelay * 2);
}
calculateDelay(attempt, error) {
// 基础退避
let delay = this.config.baseDelay * Math.pow(this.config.backoffFactor, attempt);
// 添加抖动
const jitter = 1 + (Math.random() * 2 - 1) * this.config.jitter;
delay *= jitter;
// 基于错误类型调整
if (error && error.name === 'TimeoutError') {
delay *= 1.5; // 超时错误增加延迟
}
// 基于网络质量调整
if (this.networkQuality.reliability < 0.5) {
delay *= 2; // 网络不可靠时增加延迟
}
return Math.min(delay, this.config.maxDelay);
}
shouldRetry(error, attempt) {
// 基于错误类型决定是否重试
const retryableErrors = [
'TimeoutError',
'TypeError', // 网络错误
'AbortError'
];
if (retryableErrors.includes(error.name)) {
return true;
}
// 基于HTTP状态码决定是否重试
if (error.statusCode) {
const retryableStatusCodes = [408, 429, 500, 502, 503, 504];
return retryableStatusCodes.includes(error.statusCode);
}
// 基于尝试次数决定是否重试
if (attempt >= this.config.maxRetries) {
return false;
}
// 默认不重试
return false;
}
async handleRetry(context, attempt, error) {
// 计算延迟
const delay = this.calculateDelay(attempt, error);
context.delays.push(delay);
// 记录重试
this.logRetryAttempt(context, attempt, delay, error);
// 等待延迟
await this.sleep(delay);
}
async executeFallback(failureType, context, options, error) {
const fallbackStrategy = this.config.fallbackStrategies[failureType] ||
this.config.fallbackStrategies.default;
switch (fallbackStrategy) {
case 'return_cached':
return this.returnCachedResponse(context.url, options);
case 'retry_with_exponential_backoff':
// 已在上层处理
throw error;
case 'use_alternative_endpoint':
return this.useAlternativeEndpoint(context.url, options);
case 'return_degraded_response':
return this.returnDegradedResponse(context, options);
case 'notify_user':
return this.notifyUserAndFail(context, error);
default:
throw error;
}
}
async returnCachedResponse(url, options) {
// 检查缓存
const cacheKey = this.getCacheKey(url, options);
const cached = this.fallbackCache.get(cacheKey);
if (cached && Date.now() - cached.timestamp < 300000) { // 5分钟缓存
console.log(`Using cached response for ${url}`);
// 创建模拟响应
return new Response(JSON.stringify(cached.data), {
status: 200,
headers: {
'Content-Type': 'application/json',
'X-Cached-Response': 'true',
'X-Cache-Age': Math.round((Date.now() - cached.timestamp) / 1000) + 's'
}
});
}
throw new Error('No cached response available');
}
async useAlternativeEndpoint(originalUrl, options) {
// 生成替代端点
const alternativeUrl = this.generateAlternativeUrl(originalUrl);
if (alternativeUrl && alternativeUrl !== originalUrl) {
console.log(`Trying alternative endpoint: ${alternativeUrl}`);
// 使用替代端点重试（最多一次）
return this.fetchWithRetry(alternativeUrl, {
...options,
headers: {
...options.headers,
'X-Original-URL': originalUrl
}
});
}
throw new Error('No alternative endpoint available');
}
generateAlternativeUrl(url) {
const urlObj = new URL(url);
// 尝试不同的策略
const strategies = [
// 1. 使用备用域名
() => {
if (urlObj.hostname === 'api.example.com') {
urlObj.hostname = 'api-backup.example.com';
return urlObj.toString();
}
return null;
},
// 2. 使用不同端口
() => {
if (!urlObj.port) {
urlObj.port = '8080';
return urlObj.toString();
}
return null;
},
// 3. 使用不同API版本
() => {
if (urlObj.pathname.includes('/v1/')) {
return urlObj.toString().replace('/v1/', '/v2/');
}
return null;
},
// 4. 使用HTTP代替HTTPS（仅限开发环境）
() => {
if (urlObj.protocol === 'https:' && window.location.hostname === 'localhost') {
urlObj.protocol = 'http:';
return urlObj.toString();
}
return null;
}
];
for (const strategy of strategies) {
const result = strategy();
if (result) {
return result;
}
}
return null;
}
returnDegradedResponse(context, options) {
// 返回降级响应
const degradedData = {
data: null,
warning: {
code: 'DEGRADED_SERVICE',
message: 'Service is temporarily degraded. Some features may be unavailable.',
request_id: context.requestId,
original_url: context.url
},
metadata: {
degraded: true,
timestamp: new Date().toISOString(),
attempts: context.attempt
}
};
return new Response(JSON.stringify(degradedData), {
status: 200,
headers: {
'Content-Type': 'application/json',
'X-Degraded-Service': 'true'
}
});
}
notifyUserAndFail(context, error) {
// 显示用户通知
this.showUserNotification({
type: 'error',
title: 'Connection Error',
message: 'Unable to connect to the server. Please check your connection and try again.',
details: {
requestId: context.requestId,
url: context.url,
error: error.message
}
});
throw error;
}
// 熔断器相关方法
getCircuitKey(url, method) {
const urlObj = new URL(url);
return `${method}:${urlObj.hostname}:${urlObj.pathname.split('/')[1] || ''}`;
}
allowRequest(circuitKey) {
if (!this.circuitBreakers.has(circuitKey)) {
this.circuitBreakers.set(circuitKey, {
state: 'closed',
failureCount: 0,
successCount: 0,
lastFailure: null,
openedAt: null,
halfOpenAttempts: 0
});
return true;
}
const cb = this.circuitBreakers.get(circuitKey);
if (cb.state === 'open') {
// 检查是否应该进入半开状态
if (Date.now() - cb.openedAt >= this.config.circuitBreaker.resetTimeout) {
cb.state = 'half_open';
cb.halfOpenAttempts = 0;
return true;
}
return false;
}
if (cb.state === 'half_open') {
if (cb.halfOpenAttempts >= this.config.circuitBreaker.halfOpenMaxRequests) {
return false;
}
cb.halfOpenAttempts++;
return true;
}
return true; // closed状态
}
recordSuccess(circuitKey) {
const cb = this.circuitBreakers.get(circuitKey);
if (cb) {
cb.successCount++;
cb.failureCount = Math.max(0, cb.failureCount - 1); // 衰减失败计数
if (cb.state === 'half_open' && cb.successCount >= 3) {
cb.state = 'closed';
cb.failureCount = 0;
cb.successCount = 0;
}
}
}
recordFailure(circuitKey, error) {
const cb = this.circuitBreakers.get(circuitKey);
if (cb) {
cb.failureCount++;
cb.lastFailure = Date.now();
// 检查是否应该打开熔断器
if (cb.state === 'closed' &&
cb.failureCount >= this.config.circuitBreaker.failureThreshold) {
cb.state = 'open';
cb.openedAt = Date.now();
this.logCircuitEvent(circuitKey, 'opened', error);
}
}
}
// 网络监控
initNetworkMonitoring() {
// 定期检查网络质量
setInterval(() => this.checkNetworkQuality(), 30000);
// 监听在线/离线事件
window.addEventListener('online', () => this.handleNetworkOnline());
window.addEventListener('offline', () => this.handleNetworkOffline());
}
async checkNetworkQuality() {
try {
// 测量到已知服务器的延迟
const latency = await this.measureLatency();
// 测量吞吐量
const throughput = await this.measureThroughput();
// 计算可靠性
const recentRequests = Array.from(this.requestHistory.values())
.slice(-20); // 最近20个请求
const successfulRequests = recentRequests.filter(r => r.success);
const reliability = recentRequests.length > 0 ?
successfulRequests.length / recentRequests.length : 1.0;
this.networkQuality = {
latency,
throughput,
reliability
};
console.log(`Network quality updated:`, this.networkQuality);
} catch (error) {
console.warn('Network quality check failed:', error);
}
}
async measureLatency() {
const startTime = Date.now();
try {
await fetch('/ping', {
method: 'HEAD',
cache: 'no-store'
});
return Date.now() - startTime;
} catch {
return 1000; // 默认高延迟
}
}
async measureThroughput() {
// 简单测试：下载小文件测量速度
const testSize = 10000; // 10KB
const startTime = Date.now();
try {
const response = await fetch(`/test-data?size=${testSize}`);
const blob = await response.blob();
const duration = Date.now() - startTime;
return duration > 0 ? (testSize * 8) / (duration / 1000) : 0; // bps
} catch {
return 0;
}
}
updateNetworkQuality(context, success) {
// 记录请求历史
this.requestHistory.set(context.requestId, {
url: context.url,
success,
duration: Date.now() - context.startTime,
timestamp: Date.now(),
attempts: context.attempt
});
// 保持历史记录大小
if (this.requestHistory.size > 100) {
const oldestKey = this.requestHistory.keys().next().value;
this.requestHistory.delete(oldestKey);
}
}
handleNetworkOnline() {
console.log('Network came online, resetting circuit breakers');
// 重置所有熔断器
for (const cb of this.circuitBreakers.values()) {
if (cb.state === 'open') {
cb.state = 'half_open';
cb.halfOpenAttempts = 0;
}
}
}
handleNetworkOffline() {
console.log('Network went offline');
this.networkQuality.reliability = 0;
}
// 工具方法
generateRequestId() {
return `req_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
}
getCacheKey(url, options) {
const urlObj = new URL(url);
return `${urlObj.pathname}?${urlObj.searchParams.toString()}:${options.method || 'GET'}`;
}
sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}
logRetryAttempt(context, attempt, delay, error) {
console.log(`Retry attempt ${attempt + 1} for ${context.url}, delay: ${delay}ms`, error);
}
logCircuitEvent(circuitKey, event, error) {
console.log(`Circuit breaker ${circuitKey} ${event}`, error);
}
showUserNotification(notification) {
// 在实际应用中，这里会显示UI通知
console.log('User notification:', notification);
}
getMetrics() {
return {
...this.metrics,
circuitBreakers: {
total: this.circuitBreakers.size,
open: Array.from(this.circuitBreakers.values()).filter(cb => cb.state === 'open').length,
halfOpen: Array.from(this.circuitBreakers.values()).filter(cb => cb.state === 'half_open').length
},
networkQuality: this.networkQuality
};
}
}
// 使用示例
const client = new AdaptiveRetryClient({
maxRetries: 3,
timeout: 10000,
fallbackStrategies: {
timeout: 'return_cached',
network_error: 'use_alternative_endpoint',
server_error: 'return_degraded_response'
}
});
// 发起请求
async function fetchUserData(userId) {
try {
const response = await client.fetchWithRetry(`/api/users/${userId}`, {
headers: {
'Authorization': `Bearer ${token}`
}
});
if (response.headers.get('X-Degraded-Service')) {
// 处理降级响应
return handleDegradedResponse(await response.json());
}
return await response.json();
} catch (error) {
console.error('Failed to fetch user data:', error);
return null;
}
}
21.5 监控与分析
21.5.1 超时监控与根因分析系统
python
# 超时监控与根因分析系统
from datetime import datetime, timedelta
from collections import defaultdict, deque
import statistics
from typing import Dict, List, Optional, Tuple
import numpy as np
from scipy import stats
class TimeoutRootCauseAnalyzer:
"""超时根因分析系统"""
def __init__(self, config: Dict = None):
self.config = {
'window_size': 1000, # 分析窗口大小
'percentile_threshold': 95, # 百分位阈值
'anomaly_threshold': 3.0, # 异常阈值（标准差）
'correlation_threshold': 0.7, # 相关性阈值
'min_samples': 50, # 最小样本数
** (config or {})
}
# 数据存储
self.timeout_data = deque(maxlen=self.config['window_size'])
self.normal_data = deque(maxlen=self.config['window_size'])
self.metrics_history = defaultdict(lambda: deque(maxlen=100))
# 统计信息
self.statistics = {
'total_timeouts': 0,
'timeout_rate': 0.0,
'avg_timeout_duration': 0.0,
'timeout_patterns': defaultdict(int)
}
# 根因分析模型
self.cause_models = self.initialize_cause_models()
def initialize_cause_models(self) -> Dict:
"""初始化根因分析模型"""
return {
'network': {
'indicators': ['latency', 'packet_loss', 'throughput'],
'thresholds': {
'latency': 500, # ms
'packet_loss': 0.1, # 10%
'throughput': 1024 # 1KB/s
},
'weight': 0.3
},
'server': {
'indicators': ['cpu_usage', 'memory_usage', 'disk_io'],
'thresholds': {
'cpu_usage': 0.8, # 80%
'memory_usage': 0.9, # 90%
'disk_io': 1000 # 1000 IOPS
},
'weight': 0.4
},
'application': {
'indicators': ['response_time', 'error_rate', 'queue_length'],
'thresholds': {
'response_time': 5.0, # 5秒
'error_rate': 0.05, # 5%
'queue_length': 100
},
'weight': 0.2
},
'client': {
'indicators': ['request_size', 'concurrent_requests', 'geolocation'],
'thresholds': {
'request_size': 10 * 1024 * 1024, # 10MB
'concurrent_requests': 10,
'geolocation_distance': 1000 # 1000km
},
'weight': 0.1
}
}
def record_timeout(self, timeout_event: Dict):
"""记录超时事件"""
self.timeout_data.append({
'timestamp': datetime.utcnow(),
'data': timeout_event
})
self.statistics['total_timeouts'] += 1
# 更新统计
self.update_statistics()
# 分析根因
root_cause = self.analyze_root_cause(timeout_event)
# 记录模式
if root_cause:
self.statistics['timeout_patterns'][root_cause['primary_cause']] += 1
return root_cause
def record_normal_request(self, request_data: Dict):
"""记录正常请求"""
self.normal_data.append({
'timestamp': datetime.utcnow(),
'data': request_data
})
# 更新指标历史
for metric, value in request_data.get('metrics', {}).items():
self.metrics_history[metric].append(value)
def analyze_root_cause(self, timeout_event: Dict) -> Optional[Dict]:
"""分析超时根因"""
# 收集指标
indicators = self.collect_indicators(timeout_event)
# 计算各原因的可能性
cause_scores = {}
for cause_name, cause_model in self.cause_models.items():
score = self.calculate_cause_score(cause_model, indicators)
cause_scores[cause_name] = score
# 找出主要原因
primary_cause = max(cause_scores.items(), key=lambda x: x[1])
if primary_cause[1] > 0.5: # 阈值
return {
'primary_cause': primary_cause[0],
'confidence': primary_cause[1],
'cause_scores': cause_scores,
'indicators': indicators,
'recommendations': self.generate_recommendations(primary_cause[0], indicators)
}
return None
def collect_indicators(self, timeout_event: Dict) -> Dict:
"""收集指标"""
indicators = {}
# 网络指标
indicators['latency'] = timeout_event.get('latency', 0)
indicators['packet_loss'] = timeout_event.get('packet_loss', 0)
indicators['throughput'] = timeout_event.get('throughput', 0)
# 服务器指标
indicators['cpu_usage'] = timeout_event.get('cpu_usage', 0)
indicators['memory_usage'] = timeout_event.get('memory_usage', 0)
indicators['disk_io'] = timeout_event.get('disk_io', 0)
# 应用指标
indicators['response_time'] = timeout_event.get('response_time', 0)
indicators['error_rate'] = timeout_event.get('error_rate', 0)
indicators['queue_length'] = timeout_event.get('queue_length', 0)
# 客户端指标
indicators['request_size'] = timeout_event.get('request_size', 0)
indicators['concurrent_requests'] = timeout_event.get('concurrent_requests', 0)
indicators['geolocation_distance'] = timeout_event.get('geolocation_distance', 0)
return indicators
def calculate_cause_score(self, cause_model: Dict, indicators: Dict) -> float:
"""计算原因分数"""
score = 0.0
total_weight = 0.0
for indicator in cause_model['indicators']:
if indicator in indicators:
value = indicators[indicator]
threshold = cause_model['thresholds'].get(indicator, 0)
# 计算指标分数
indicator_score = self.calculate_indicator_score(value, threshold)
# 加权
score += indicator_score
total_weight += 1.0
if total_weight > 0:
score = score / total_weight * cause_model['weight']
return score
def calculate_indicator_score(self, value: float, threshold: float) -> float:
"""计算指标分数"""
if threshold == 0:
return 0.0
# 归一化并计算超出程度
normalized = min(value / threshold, 2.0) # 限制在2倍以内
score = max(0, normalized - 1) # 超过阈值部分
return score
def generate_recommendations(self, cause: str, indicators: Dict) -> List[str]:
"""生成推荐解决方案"""
recommendations = []
if cause == 'network':
if indicators.get('latency', 0) > 500:
recommendations.append('优化网络路由或使用CDN')
if indicators.get('packet_loss', 0) > 0.1:
recommendations.append('检查网络稳定性或增加重试机制')
if indicators.get('throughput', 0) < 1024:
recommendations.append('升级网络带宽或压缩传输数据')
elif cause == 'server':
if indicators.get('cpu_usage', 0) > 0.8:
recommendations.append('增加CPU资源或优化代码')
if indicators.get('memory_usage', 0) > 0.9:
recommendations.append('增加内存或优化内存使用')
if indicators.get('disk_io', 0) > 1000:
recommendations.append('优化磁盘I/O或使用SSD')
elif cause == 'application':
if indicators.get('response_time', 0) > 5.0:
recommendations.append('优化数据库查询或缓存策略')
if indicators.get('error_rate', 0) > 0.05:
recommendations.append('修复应用错误或增加错误处理')
if indicators.get('queue_length', 0) > 100:
recommendations.append('增加处理能力或实现负载均衡')
elif cause == 'client':
if indicators.get('request_size', 0) > 10 * 1024 * 1024:
recommendations.append('压缩请求数据或分块上传')
if indicators.get('concurrent_requests', 0) > 10:
recommendations.append('限制客户端并发请求数')
# 通用推荐
recommendations.extend([
'增加超时时间配置',
'实现更好的重试机制',
'监控系统性能指标'
])
return recommendations
def detect_anomalies(self) -> List[Dict]:
"""检测异常模式"""
anomalies = []
if len(self.timeout_data) < self.config['min_samples']:
return anomalies
# 分析超时率
timeout_rate = self.calculate_timeout_rate()
if timeout_rate > self.calculate_threshold('timeout_rate'):
anomalies.append({
'type': 'high_timeout_rate',
'value': timeout_rate,
'threshold': self.calculate_threshold('timeout_rate'),
'severity': 'high'
})
# 分析超时持续时间
durations = [event['data'].get('duration', 0) for event in self.timeout_data]
if durations:
avg_duration = statistics.mean(durations)
p95_duration = np.percentile(durations, 95)
if p95_duration > self.calculate_threshold('timeout_duration'):
anomalies.append({
'type': 'long_timeout_duration',
'p95_duration': p95_duration,
'threshold': self.calculate_threshold('timeout_duration'),
'severity': 'medium'
})
# 分析时间相关性
time_pattern = self.analyze_time_pattern()
if time_pattern['has_pattern']:
anomalies.append({
'type': 'time_pattern_detected',
'pattern': time_pattern['pattern'],
'confidence': time_pattern['confidence'],
'severity': 'low'
})
return anomalies
def calculate_timeout_rate(self) -> float:
"""计算超时率"""
window_seconds = 300 # 5分钟窗口
cutoff = datetime.utcnow() - timedelta(seconds=window_seconds)
timeout_count = sum(1 for event in self.timeout_data
if event['timestamp'] > cutoff)
total_count = timeout_count + len([event for event in self.normal_data
if event['timestamp'] > cutoff])
return timeout_count / total_count if total_count > 0 else 0.0
def calculate_threshold(self, metric: str) -> float:
"""计算异常阈值"""
# 基于历史数据计算动态阈值
if metric == 'timeout_rate':
historical_values = [self.statistics.get('timeout_rate', 0)]
return np.mean(historical_values) + 2 * np.std(historical_values)
elif metric == 'timeout_duration':
durations = [event['data'].get('duration', 0) for event in self.timeout_data]
if durations:
return np.percentile(durations, 95) * 1.5
return 30.0 # 默认30秒
return 0.0
def analyze_time_pattern(self) -> Dict:
"""分析时间模式"""
if len(self.timeout_data) < 10:
return {'has_pattern': False}
# 提取超时时间点
time_points = [event['timestamp'].hour for event in self.timeout_data]
# 分析时间分布
hour_counts = defaultdict(int)
for hour in time_points:
hour_counts[hour] += 1
# 检查是否有明显的时间模式
max_hour = max(hour_counts.items(), key=lambda x: x[1])
total = len(time_points)
if max_hour[1] / total > 0.3: # 30%的超时发生在同一小时
return {
'has_pattern': True,
'pattern': f'Peak at hour {max_hour[0]}',
'confidence': max_hour[1] / total
}
return {'has_pattern': False}
def update_statistics(self):
"""更新统计信息"""
window_seconds = 3600 # 1小时窗口
cutoff = datetime.utcnow() - timedelta(seconds=window_seconds)
# 计算超时率
timeout_count = sum(1 for event in self.timeout_data
if event['timestamp'] > cutoff)
total_count = timeout_count + len([event for event in self.normal_data
if event['timestamp'] > cutoff])
self.statistics['timeout_rate'] = timeout_count / total_count if total_count > 0 else 0.0
# 计算平均超时持续时间
recent_timeouts = [event for event in self.timeout_data
if event['timestamp'] > cutoff]
if recent_timeouts:
durations = [event['data'].get('duration', 0) for event in recent_timeouts]
self.statistics['avg_timeout_duration'] = statistics.mean(durations)
def generate_report(self, period: str = '24h') -> Dict:
"""生成分析报告"""
if period == '24h':
cutoff = datetime.utcnow() - timedelta(hours=24)
elif period == '7d':
cutoff = datetime.utcnow() - timedelta(days=7)
else:
cutoff = datetime.utcnow() - timedelta(hours=24)
# 筛选期间数据
period_timeouts = [event for event in self.timeout_data
if event['timestamp'] > cutoff]
period_normals = [event for event in self.normal_data
if event['timestamp'] > cutoff]
# 基础统计
total_requests = len(period_timeouts) + len(period_normals)
timeout_rate = len(period_timeouts) / total_requests if total_requests > 0 else 0
# 根因分布
cause_distribution = defaultdict(int)
for event in period_timeouts:
root_cause = self.analyze_root_cause(event['data'])
if root_cause:
cause_distribution[root_cause['primary_cause']] += 1
# 时间分布
hourly_distribution = defaultdict(int)
for event in period_timeouts:
hour = event['timestamp'].hour
hourly_distribution[f'{hour:02d}:00'] += 1
report = {
'period': period,
'time_range': {
'start': cutoff.isoformat(),
'end': datetime.utcnow().isoformat()
},
'summary': {
'total_requests': total_requests,
'timeout_count': len(period_timeouts),
'timeout_rate': timeout_rate,
'avg_timeout_duration': self.statistics['avg_timeout_duration']
},
'cause_analysis': {
'distribution': dict(cause_distribution),
'top_causes': sorted(cause_distribution.items(),
key=lambda x: x[1], reverse=True)[:5]
},
'time_distribution': dict(sorted(hourly_distribution.items())),
'anomalies': self.detect_anomalies(),
'recommendations': self.generate_overall_recommendations()
}
return report
def generate_overall_recommendations(self) -> List[str]:
"""生成整体推荐"""
recommendations = []
# 基于根因分析
for cause, count in self.statistics['timeout_patterns'].items():
if count > 10: # 频繁出现的原因
if cause == 'network':
recommendations.append('考虑优化网络基础设施或使用CDN')
elif cause == 'server':
recommendations.append('建议增加服务器资源或优化配置')
elif cause == 'application':
recommendations.append('需要性能优化和代码审查')
elif cause == 'client':
recommendations.append('建议改善客户端实现和添加限制')
# 基于超时率
if self.statistics['timeout_rate'] > 0.1: # 超时率超过10%
recommendations.append('整体超时率过高，建议全面审查系统架构')
# 基于持续时间
if self.statistics['avg_timeout_duration'] > 30: # 平均超时超过30秒
recommendations.append('超时持续时间过长，建议增加超时配置或优化处理逻辑')
return list(set(recommendations))[:5] # 去重并限制数量
21.5.2 实时超时监控仪表板
javascript
// React组件：实时超时监控仪表板
import React, { useState, useEffect, useRef } from 'react';
import {
LineChart, Line, BarChart, Bar, PieChart, Pie, Cell,
XAxis, YAxis, CartesianGrid, Tooltip, Legend, ResponsiveContainer,
AreaChart, Area, RadarChart, Radar, PolarGrid, PolarAngleAxis, PolarRadiusAxis
} from 'recharts';
import { AlertTriangle, Clock, Activity, Wifi, Server, Cpu, AlertCircle } from 'lucide-react';
function TimeoutMonitoringDashboard() {
const [timeRange, setTimeRange] = useState('1h');
const [stats, setStats] = useState(null);
const [realtimeEvents, setRealtimeEvents] = useState([]);
const [anomalies, setAnomalies] = useState([]);
const [causeDistribution, setCauseDistribution] = useState([]);
const [networkMetrics, setNetworkMetrics] = useState({});
const wsRef = useRef(null);
// 颜色配置
const COLORS = ['#0088FE', '#00C49F', '#FFBB28', '#FF8042', '#8884D8'];
// 获取统计数据
useEffect(() => {
const fetchStats = async () => {
try {
const response = await fetch(`/api/monitor/timeout-stats?range=${timeRange}`);
const data = await response.json();
setStats(data);
// 处理根因分布数据
if (data.cause_analysis?.distribution) {
const distribution = Object.entries(data.cause_analysis.distribution)
.map(([name, value]) => ({ name, value }));
setCauseDistribution(distribution);
}
} catch (error) {
console.error('Failed to fetch stats:', error);
}
};
fetchStats();
const interval = setInterval(fetchStats, 30000); // 每30秒更新
return () => clearInterval(interval);
}, [timeRange]);
// WebSocket连接接收实时事件
useEffect(() => {
const connectWebSocket = () => {
wsRef.current = new WebSocket('wss://api.example.com/monitor/timeout-events');
wsRef.current.onmessage = (event) => {
const eventData = JSON.parse(event.data);
// 添加事件到实时列表
setRealtimeEvents(prev => [
{ ...eventData, id: Date.now() },
...prev.slice(0, 49) // 保留最近50个
]);
// 检查是否是异常事件
if (eventData.severity === 'high') {
setAnomalies(prev => [
{
...eventData,
id: Date.now(),
acknowledged: false
},
...prev
]);
}
// 更新网络指标
if (eventData.metrics) {
setNetworkMetrics(prev => ({
...prev,
...eventData.metrics,
lastUpdated: Date.now()
}));
}
};
wsRef.current.onclose = () => {
console.log('WebSocket closed, reconnecting...');
setTimeout(connectWebSocket, 5000);
};
wsRef.current.onerror = (error) => {
console.error('WebSocket error:', error);
};
};
connectWebSocket();
return () => {
if (wsRef.current) {
wsRef.current.close();
}
};
}, []);
// 模拟网络指标更新
useEffect(() => {
const updateNetworkMetrics = () => {
setNetworkMetrics(prev => ({
latency: Math.random() * 500,
throughput: 1000 + Math.random() * 5000,
packetLoss: Math.random() * 0.05,
reliability: 0.95 + Math.random() * 0.05,
lastUpdated: Date.now()
}));
};
const interval = setInterval(updateNetworkMetrics, 10000);
return () => clearInterval(interval);
}, []);
if (!stats) {
return (
加载监控数据...
);
}
return (
{/* 仪表板头部 */}
请求超时监控
时间范围: {timeRange === '1h' ? '1小时' :
timeRange === '24h' ? '24小时' :
timeRange === '7d' ? '7天' : '30天'}
{['1h', '24h', '7d', '30d'].map(range => (
))}
{/* 关键指标卡片 */}
超时总数
{stats.summary.timeout_count}
率: {(stats.summary.timeout_rate * 100).toFixed(2)}%
平均超时
{stats.summary.avg_timeout_duration?.toFixed(1) || '0.0'}s
阈值: 30.0s
总请求数
{stats.summary.total_requests?.toLocaleString() || '0'}
{stats.summary.total_requests > 0 ? '正常' : '无数据'}
正常率
{((1 - stats.summary.timeout_rate) * 100).toFixed(1)}%
目标: 99.9%
{/* 网络指标卡片 */}
网络质量指标
超时趋势
({ hour, count }))}>
{ r: 4 }}
activeDot={{ r: 6 }}
name="超时次数"
/>
根因分布
`${name}: ${(percent * 100).toFixed(0)}%`}
outerRadius={80}
fill="#8884d8"
dataKey="value"
>
{causeDistribution.map((entry, index) => (
|
))}
[`${value}次`, '数量']} />
超时率变化
{ value: '超时率 (%)', angle: -90, position: 'insideLeft' }} />
[`${value}%`, '超时率']} />
性能指标雷达图
{/* 实时事件流 */}
实时超时事件
时间
端点
持续时间
根因
客户端IP
严重程度
操作
{realtimeEvents.map(event => (
{new Date(event.timestamp || Date.now()).toLocaleTimeString()}
{event.endpoint?.length > 30 ?
event.endpoint.substring(0, 30) + '...' :
event.endpoint}
{event.duration ? `${event.duration.toFixed(2)}s` : 'N/A'}
{event.root_cause || '未知'}
{event.client_ip || 'N/A'}
{event.severity === 'high' ? '高' :
event.severity === 'medium' ? '中' : '低'}
))}
{/* 异常警报 */}
{anomalies.length > 0 && (
异常警报 ({anomalies.length})
{anomalies.filter(a => !a.acknowledged).slice(0, 5).map(anomaly => (
{anomaly.type}
{new Date(anomaly.timestamp).toLocaleTimeString()}
{anomaly.message || '检测到异常模式'}
{anomaly.details && (
{JSON.stringify(anomaly.details, null, 2)}
)}
))}
)}
{/* 建议和洞察 */}
优化建议
{stats.recommendations?.map((recommendation, index) => (
💡
{recommendation}
))}
);
// 辅助函数
function generateRateData() {
// 生成模拟的超时率数据
const data = [];
for (let i = 0; i < 24; i++) {
data.push({
time: `${i}:00`,
rate: Math.random() * 15, // 0-15%的超时率
threshold: 10 // 警告阈值10%
});
}
return data;
}
function generateRadarData() {
// 生成雷达图数据
return [
{ subject: '网络质量', current: 85, target: 95 },
{ subject: '服务器性能', current: 90, target: 98 },
{ subject: '应用响应', current: 75, target: 95 },
{ subject: '客户端体验', current: 80, target: 90 },
{ subject: '系统稳定性', current: 88, target: 99 },
{ subject: '错误恢复', current: 70, target: 85 }
];
}
function viewEventDetails(event) {
// 查看事件详情
console.log('Event details:', event);
alert(`事件详情:
${JSON.stringify(event, null, 2)}`);
}
function acknowledgeEvent(eventId) {
// 确认事件
setRealtimeEvents(prev =>
prev.map(event =>
event.id === eventId ? { ...event, acknowledged: true } : event
)
);
}
function acknowledgeAnomaly(anomalyId) {
// 确认异常
setAnomalies(prev =>
prev.map(anomaly =>
anomaly.id === anomalyId ? { ...anomaly, acknowledged: true } : anomaly
)
);
}
function investigateAnomaly(anomaly) {
// 调查异常
console.log('Investigating anomaly:', anomaly);
window.open(`/monitor/investigate?anomalyId=${anomaly.id}`, '_blank');
}
function muteAnomaly(anomalyId) {
// 静音异常
setAnomalies(prev => prev.filter(a => a.id !== anomalyId));
}
function implementRecommendation(recommendation) {
// 实施建议
console.log('Implementing recommendation:', recommendation);
alert(`开始实施: ${recommendation}`);
}
function exportReport() {
// 导出报告
const reportData = {
timestamp: new Date().toISOString(),
timeRange,
stats,
networkMetrics
};
const blob = new Blob([JSON.stringify(reportData, null, 2)], {
type: 'application/json'
});
const url = URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
a.download = `timeout-report-${new Date().toISOString().slice(0, 10)}.json`;
a.click();
URL.revokeObjectURL(url);
}
}
// CSS样式（内联示例）
const styles = `
.timeout-dashboard {
padding: 20px;
max-width: 1600px;
margin: 0 auto;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
}
.dashboard-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 30px;
padding-bottom: 20px;
border-bottom: 1px solid #e0e0e0;
}
.header-left {
display: flex;
align-items: center;
gap: 15px;
}
.header-icon {
color: #ff6b6b;
}
.time-range-label {
background: #f0f0f0;
padding: 4px 12px;
border-radius: 16px;
font-size: 14px;
color: #666;
}
.time-range-selector {
display: flex;
gap: 8px;
}
.time-range-btn {
padding: 8px 16px;
border: 1px solid #ddd;
border-radius: 4px;
background: white;
cursor: pointer;
font-size: 14px;
}
.time-range-btn.active {
background: #007bff;
color: white;
border-color: #007bff;
}
.key-metrics-cards {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(250px, 1fr));
gap: 20px;
margin-bottom: 30px;
}
.metric-card {
background: white;
border-radius: 8px;
padding: 20px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
display: flex;
align-items: center;
gap: 20px;
border-left: 4px solid;
}
.metric-card.critical { border-left-color: #ff6b6b; }
.metric-card.warning { border-left-color: #ffa726; }
.metric-card.info { border-left-color: #42a5f5; }
.metric-card.success { border-left-color: #66bb6a; }
.metric-icon {
background: #f5f5f5;
padding: 12px;
border-radius: 8px;
display: flex;
align-items: center;
justify-content: center;
}
.metric-card.critical .metric-icon { background: #ffebee; color: #ff6b6b; }
.metric-card.warning .metric-icon { background: #fff3e0; color: #ffa726; }
.metric-card.info .metric-icon { background: #e3f2fd; color: #42a5f5; }
.metric-card.success .metric-icon { background: #e8f5e8; color: #66bb6a; }
.metric-value {
font-size: 32px;
font-weight: bold;
margin: 8px 0;
}
.metric-trend {
font-size: 14px;
color: #666;
}
.network-metrics {
background: white;
border-radius: 8px;
padding: 20px;
margin-bottom: 30px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
}
.network-cards {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
gap: 15px;
margin-top: 15px;
}
.network-card {
display: flex;
align-items: center;
gap: 10px;
padding: 15px;
background: #f8f9fa;
border-radius: 6px;
border: 1px solid #e9ecef;
}
.network-label {
flex: 1;
font-size: 14px;
color: #666;
}
.network-value {
font-weight: bold;
font-size: 16px;
}
.network-value.warning { color: #ffa726; }
.network-value.critical { color: #ff6b6b; }
.charts-section {
margin-bottom: 30px;
}
.chart-row {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(600px, 1fr));
gap: 20px;
margin-bottom: 20px;
}
.chart-container {
background: white;
border-radius: 8px;
padding: 20px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
}
.realtime-events-section {
background: white;
border-radius: 8px;
padding: 20px;
margin-bottom: 30px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
}
.events-table-container {
overflow-x: auto;
margin-top: 15px;
}
.events-table {
width: 100%;
border-collapse: collapse;
}
.events-table th {
text-align: left;
padding: 12px;
background: #f8f9fa;
border-bottom: 2px solid #dee2e6;
font-weight: 600;
color: #495057;
}
.events-table td {
padding: 12px;
border-bottom: 1px solid #dee2e6;
}
.event-row.severity-high {
background: #fff5f5;
}
.event-row.severity-medium {
background: #fff9db;
}
.event-row.severity-low {
background: #f8f9fa;
}
.endpoint-cell code {
background: #f1f3f5;
padding: 2px 6px;
border-radius: 4px;
font-family: monospace;
font-size: 12px;
}
.duration-badge {
background: #e7f5ff;
color: #1971c2;
padding: 4px 8px;
border-radius: 4px;
font-size: 12px;
font-weight: 500;
}
.cause-badge {
padding: 4px 8px;
border-radius: 4px;
font-size: 12px;
font-weight: 500;
}
.cause-badge.cause-network { background: #e3f2fd; color: #1976d2; }
.cause-badge.cause-server { background: #f3e5f5; color: #7b1fa2; }
.cause-badge.cause-application { background: #e8f5e8; color: #2e7d32; }
.cause-badge.cause-client { background: #fff3e0; color: #ef6c00; }
.severity-badge {
padding: 4px 8px;
border-radius: 4px;
font-size: 12px;
font-weight: 500;
}
.severity-badge.severity-high { background: #ffebee; color: #c62828; }
.severity-badge.severity-medium { background: #fff3e0; color: #ef6c00; }
.severity-badge.severity-low { background: #e8f5e8; color: #2e7d32; }
.action-btn {
padding: 6px 12px;
margin: 0 4px;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 12px;
}
.action-btn.view-details {
background: #e3f2fd;
color: #1976d2;
}
.action-btn.acknowledge {
background: #e8f5e8;
color: #2e7d32;
}
.anomalies-section {
background: white;
border-radius: 8px;
padding: 20px;
margin-bottom: 30px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
border: 2px solid #ffebee;
}
.anomaly-card {
background: #fff5f5;
border: 1px solid #ffcdd2;
border-radius: 6px;
padding: 15px;
margin-bottom: 10px;
}
.anomaly-header {
display: flex;
justify-content: space-between;
margin-bottom: 10px;
}
.anomaly-type {
font-weight: bold;
color: #c62828;
}
.anomaly-time {
font-size: 12px;
color: #666;
}
.anomaly-details {
background: white;
padding: 10px;
border-radius: 4px;
margin-top: 10px;
font-size: 12px;
max-height: 200px;
overflow-y: auto;
}
.anomaly-actions {
display: flex;
gap: 10px;
margin-top: 15px;
}
.btn {
padding: 8px 16px;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 14px;
}
.btn-primary {
background: #1976d2;
color: white;
}
.btn-secondary {
background: #f5f5f5;
color: #333;
}
.btn-tertiary {
background: transparent;
color: #666;
border: 1px solid #ddd;
}
.insights-section {
background: white;
border-radius: 8px;
padding: 20px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
}
.insights-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
gap: 20px;
margin-top: 20px;
}
.insight-card {
display: flex;
gap: 15px;
padding: 15px;
background: #f8f9fa;
border-radius: 6px;
border-left: 4px solid #42a5f5;
}
.insight-icon {
font-size: 24px;
}
.insight-content {
flex: 1;
}
.insight-action {
margin-top: 10px;
padding: 6px 12px;
background: #42a5f5;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 12px;
}
`;
// 添加到文档
const styleSheet = document.createElement('style');
styleSheet.textContent = styles;
document.head.appendChild(styleSheet)
