python<\/p>\n
# \u72b6\u6001\u7801\u54cd\u5e94\u65f6\u95f4\u5206\u6790\u5de5\u5177
\nfrom typing import Dict, List, Tuple
\nimport time
\nfrom dataclasses import dataclass
\nfrom enum import Enum
\nimport statistics<\/p>\n
class ResponseTimeComponent(Enum):
\n """\u54cd\u5e94\u65f6\u95f4\u7ec4\u4ef6"""
\n NETWORK = "network" # \u7f51\u7edc\u4f20\u8f93\u65f6\u95f4
\n DNS = "dns" # DNS\u89e3\u6790\u65f6\u95f4
\n SSL = "ssl" # TLS\u63e1\u624b\u65f6\u95f4
\n QUEUEING = "queueing" # \u8bf7\u6c42\u6392\u961f\u65f6\u95f4
\n PROCESSING = "processing" # \u670d\u52a1\u5668\u5904\u7406\u65f6\u95f4
\n DATABASE = "database" # \u6570\u636e\u5e93\u67e5\u8be2\u65f6\u95f4
\n CACHE = "cache" # \u7f13\u5b58\u8bbf\u95ee\u65f6\u95f4
\n EXTERNAL_API = "external_api" # \u5916\u90e8API\u8c03\u7528\u65f6\u95f4
\n SERIALIZATION = "serialization" # \u6570\u636e\u5e8f\u5217\u5316\u65f6\u95f4
\n UNKNOWN = "unknown" # \u672a\u77e5\u65f6\u95f4<\/p>\n
@dataclass
\nclass TimingMeasurement:
\n """\u65f6\u95f4\u6d4b\u91cf"""
\n component: ResponseTimeComponent
\n start_time: float
\n end_time: float<\/p>\n
@property
\n def duration_ms(self) -> float:
\n return (self.end_time – self.start_time) * 1000<\/p>\n
@dataclass
\nclass RequestProfile:
\n """\u8bf7\u6c42\u6027\u80fd\u5256\u6790"""
\n request_id: str
\n status_code: int
\n total_time_ms: float
\n timings: List[TimingMeasurement]
\n endpoint: str
\n method: str<\/p>\n
def get_component_time(self, component: ResponseTimeComponent) -> float:
\n """\u83b7\u53d6\u7ec4\u4ef6\u65f6\u95f4"""
\n component_timings = [
\n t for t in self.timings
\n if t.component == component
\n ]
\n return sum(t.duration_ms for t in component_timings)<\/p>\n
def get_time_breakdown(self) -> Dict[ResponseTimeComponent, float]:
\n """\u83b7\u53d6\u65f6\u95f4\u5206\u89e3"""
\n breakdown = {}
\n for component in ResponseTimeComponent:
\n time_ms = self.get_component_time(component)
\n if time_ms > 0:
\n breakdown[component] = time_ms
\n return breakdown<\/p>\n
def get_percentage_breakdown(self) -> Dict[ResponseTimeComponent, float]:
\n """\u83b7\u53d6\u767e\u5206\u6bd4\u5206\u89e3"""
\n breakdown = self.get_time_breakdown()
\n if not breakdown or self.total_time_ms == 0:
\n return {}<\/p>\n
return {
\n component: (time_ms \/ self.total_time_ms) * 100
\n for component, time_ms in breakdown.items()
\n }<\/p>\n
class PerformanceAnalyzer:
\n """\u6027\u80fd\u5206\u6790\u5668"""<\/p>\n
def __init__(self):
\n self.profiles: Dict[str, RequestProfile] = {}
\n self.status_code_stats: Dict[int, List[RequestProfile]] = {}
\n self.endpoint_stats: Dict[str, List[RequestProfile]] = {}<\/p>\n
def add_profile(self, profile: RequestProfile):
\n """\u6dfb\u52a0\u6027\u80fd\u5256\u6790"""
\n self.profiles[profile.request_id] = profile<\/p>\n
# \u6309\u72b6\u6001\u7801\u7edf\u8ba1
\n if profile.status_code not in self.status_code_stats:
\n self.status_code_stats[profile.status_code] = []
\n self.status_code_stats[profile.status_code].append(profile)<\/p>\n
# \u6309\u7aef\u70b9\u7edf\u8ba1
\n if profile.endpoint not in self.endpoint_stats:
\n self.endpoint_stats[profile.endpoint] = []
\n self.endpoint_stats[profile.endpoint].append(profile)<\/p>\n
def analyze_by_status_code(self, status_code: int) -> Dict:
\n """\u6309\u72b6\u6001\u7801\u5206\u6790\u6027\u80fd"""
\n if status_code not in self.status_code_stats:
\n return {}<\/p>\n
profiles = self.status_code_stats[status_code]
\n return self._analyze_profiles(profiles, f"status_code_{status_code}")<\/p>\n
def analyze_by_endpoint(self, endpoint: str) -> Dict:
\n """\u6309\u7aef\u70b9\u5206\u6790\u6027\u80fd"""
\n if endpoint not in self.endpoint_stats:
\n return {}<\/p>\n
profiles = self.endpoint_stats[endpoint]
\n return self._analyze_profiles(profiles, f"endpoint_{endpoint}")<\/p>\n
def _analyze_profiles(self, profiles: List[RequestProfile], group_name: str) -> Dict:
\n """\u5206\u6790\u6027\u80fd\u5256\u6790\u7ec4"""
\n if not profiles:
\n return {}<\/p>\n
total_times = [p.total_time_ms for p in profiles]<\/p>\n
# \u65f6\u95f4\u7ec4\u4ef6\u5206\u6790
\n component_times = self._analyze_components(profiles)<\/p>\n
# \u8bc6\u522b\u74f6\u9888
\n bottlenecks = self._identify_bottlenecks(component_times)<\/p>\n
# \u8d8b\u52bf\u5206\u6790
\n trends = self._analyze_trends(profiles)<\/p>\n
return {
\n 'group': group_name,
\n 'count': len(profiles),
\n 'total_time_stats': {
\n 'mean': statistics.mean(total_times),
\n 'median': statistics.median(total_times),
\n 'p95': self._calculate_percentile(total_times, 95),
\n 'p99': self._calculate_percentile(total_times, 99),
\n 'std_dev': statistics.stdev(total_times) if len(total_times) > 1 else 0,
\n 'min': min(total_times),
\n 'max': max(total_times)
\n },
\n 'component_analysis': component_times,
\n 'bottlenecks': bottlenecks,
\n 'trends': trends,
\n 'recommendations': self._generate_recommendations(bottlenecks, component_times)
\n }<\/p>\n
def _analyze_components(self, profiles: List[RequestProfile]) -> Dict:
\n """\u5206\u6790\u65f6\u95f4\u7ec4\u4ef6"""
\n component_data = {}<\/p>\n
for component in ResponseTimeComponent:
\n times = []
\n percentages = []<\/p>\n
for profile in profiles:
\n time_ms = profile.get_component_time(component)
\n if time_ms > 0:
\n times.append(time_ms)
\n percentages.append(
\n (time_ms \/ profile.total_time_ms) * 100
\n if profile.total_time_ms > 0 else 0
\n )<\/p>\n
if times:
\n component_data[component.value] = {
\n 'count': len(times),
\n 'time_stats': {
\n 'mean': statistics.mean(times),
\n 'median': statistics.median(times),
\n 'p95': self._calculate_percentile(times, 95),
\n 'p99': self._calculate_percentile(times, 99)
\n },
\n 'percentage_stats': {
\n 'mean': statistics.mean(percentages),
\n 'median': statistics.median(percentages),
\n 'p95': self._calculate_percentile(percentages, 95)
\n }
\n }<\/p>\n
return component_data<\/p>\n
def _identify_bottlenecks(self, component_analysis: Dict) -> List[Dict]:
\n """\u8bc6\u522b\u74f6\u9888"""
\n bottlenecks = []<\/p>\n
for component, data in component_analysis.items():
\n time_stats = data['time_stats']
\n percentage_stats = data['percentage_stats']<\/p>\n
# \u68c0\u67e5\u7edd\u5bf9\u65f6\u95f4\u74f6\u9888
\n if time_stats['p95'] > 1000: # \u8d85\u8fc71\u79d2
\n bottlenecks.append({
\n 'component': component,
\n 'type': 'ABSOLUTE_TIME',
\n 'value': time_stats['p95'],
\n 'threshold': 1000,
\n 'severity': 'HIGH' if time_stats['p95'] > 5000 else 'MEDIUM'
\n })<\/p>\n
# \u68c0\u67e5\u76f8\u5bf9\u65f6\u95f4\u74f6\u9888
\n if percentage_stats['mean'] > 50: # \u8d85\u8fc750%\u7684\u65f6\u95f4
\n bottlenecks.append({
\n 'component': component,
\n 'type': 'RELATIVE_TIME',
\n 'value': percentage_stats['mean'],
\n 'threshold': 50,
\n 'severity': 'HIGH' if percentage_stats['mean'] > 80 else 'MEDIUM'
\n })<\/p>\n
# \u68c0\u67e5\u9ad8\u65b9\u5dee（\u4e0d\u7a33\u5b9a\u6027）
\n time_variance = time_stats['p99'] – time_stats['mean']
\n if time_variance > time_stats['mean'] * 2: # \u65b9\u5dee\u8d85\u8fc7\u5747\u503c\u76842\u500d
\n bottlenecks.append({
\n 'component': component,
\n 'type': 'HIGH_VARIANCE',
\n 'value': time_variance,
\n 'threshold': time_stats['mean'] * 2,
\n 'severity': 'MEDIUM'
\n })<\/p>\n
# \u6309\u4e25\u91cd\u7a0b\u5ea6\u6392\u5e8f
\n severity_order = {'HIGH': 0, 'MEDIUM': 1, 'LOW': 2}
\n bottlenecks.sort(key=lambda x: severity_order[x['severity']])<\/p>\n
return bottlenecks<\/p>\n
def _analyze_trends(self, profiles: List[RequestProfile]) -> Dict:
\n """\u5206\u6790\u8d8b\u52bf"""
\n if len(profiles) < 10:
\n return {}<\/p>\n
# \u6309\u65f6\u95f4\u6392\u5e8f
\n sorted_profiles = sorted(profiles, key=lambda p: p.timings[0].start_time)<\/p>\n
# \u8ba1\u7b97\u79fb\u52a8\u5e73\u5747
\n window_size = min(10, len(sorted_profiles) \/\/ 5)
\n moving_averages = []<\/p>\n
for i in range(len(sorted_profiles) – window_size + 1):
\n window = sorted_profiles[i:i + window_size]
\n avg_time = statistics.mean(p.total_time_ms for p in window)
\n moving_averages.append(avg_time)<\/p>\n
# \u68c0\u6d4b\u8d8b\u52bf
\n trend = self._detect_trend(moving_averages)<\/p>\n
return {
\n 'sample_size': len(sorted_profiles),
\n 'moving_average_window': window_size,
\n 'trend': trend,
\n 'latest_moving_average': moving_averages[-1] if moving_averages else 0
\n }<\/p>\n
def _detect_trend(self, values: List[float]) -> str:
\n """\u68c0\u6d4b\u8d8b\u52bf"""
\n if len(values) < 2:
\n return "INSUFFICIENT_DATA"<\/p>\n
# \u4f7f\u7528\u7ebf\u6027\u56de\u5f52\u7b80\u5355\u5224\u65ad\u8d8b\u52bf
\n from scipy import stats
\n x = list(range(len(values)))
\n slope, intercept, r_value, p_value, std_err = stats.linregress(x, values)<\/p>\n
if abs(slope) < 0.1: # \u53d8\u5316\u5f88\u5c0f
\n return "STABLE"
\n elif slope > 0.5: # \u660e\u663e\u4e0a\u5347
\n return "INCREASING"
\n elif slope < -0.5: # \u660e\u663e\u4e0b\u964d
\n return "DECREASING"
\n else:
\n return "SLIGHT_CHANGE"<\/p>\n
def _generate_recommendations(self,
\n bottlenecks: List[Dict],
\n component_analysis: Dict) -> List[str]:
\n """\u751f\u6210\u4f18\u5316\u5efa\u8bae"""
\n recommendations = []<\/p>\n
for bottleneck in bottlenecks:
\n component = bottleneck['component']<\/p>\n
if component == 'database':
\n recommendations.append(
\n f"Database queries are a bottleneck ({bottleneck['value']:.0f}ms). "
\n f"Consider: 1) Adding indexes, 2) Query optimization, "
\n f"3) Implementing caching, 4) Database connection pooling."
\n )<\/p>\n
elif component == 'external_api':
\n recommendations.append(
\n f"External API calls are slow ({bottleneck['value']:.0f}ms). "
\n f"Consider: 1) Implementing timeout and retry logic, "
\n f"2) Adding circuit breaker, 3) Caching responses, "
\n f"4) Async processing if possible."
\n )<\/p>\n
elif component == 'network':
\n recommendations.append(
\n f"Network latency is high ({bottleneck['value']:.0f}ms). "
\n f"Consider: 1) Using CDN, 2) Optimizing payload size, "
\n f"3) HTTP\/2 or HTTP\/3, 4) Edge computing."
\n )<\/p>\n
elif component == 'serialization':
\n recommendations.append(
\n f"Data serialization is slow ({bottleneck['value']:.0f}ms). "
\n f"Consider: 1) Using binary formats (Protocol Buffers, MessagePack), "
\n f"2) Reducing nested structures, 3) Lazy loading."
\n )<\/p>\n
elif component == 'processing':
\n recommendations.append(
\n f"Server processing time is high ({bottleneck['value']:.0f}ms). "
\n f"Consider: 1) Code profiling and optimization, "
\n f"2) Using more efficient algorithms, "
\n f"3) Parallel processing, 4) Scaling horizontally."
\n )<\/p>\n
# \u57fa\u4e8e\u7ec4\u4ef6\u5206\u6790\u7684\u901a\u7528\u5efa\u8bae
\n if 'cache' in component_analysis:
\n cache_stats = component_analysis['cache']['percentage_stats']
\n if cache_stats['mean'] < 10: # \u7f13\u5b58\u4f7f\u7528\u7387\u4f4e
\n recommendations.append(
\n "Cache utilization is low. Consider increasing cache hit rate "
\n "through better cache strategies and cache warming."
\n )<\/p>\n
return recommendations<\/p>\n
def _calculate_percentile(self, values: List[float], percentile: float) -> float:
\n """\u8ba1\u7b97\u767e\u5206\u4f4d\u6570"""
\n if not values:
\n return 0<\/p>\n
sorted_values = sorted(values)
\n index = (percentile \/ 100) * (len(sorted_values) – 1)<\/p>\n
if index.is_integer():
\n return sorted_values[int(index)]
\n else:
\n lower = sorted_values[int(index)]
\n upper = sorted_values[int(index) + 1]
\n return lower + (upper – lower) * (index – int(index))<\/p>\n
def generate_overall_report(self) -> Dict:
\n """\u751f\u6210\u603b\u4f53\u62a5\u544a"""
\n all_profiles = list(self.profiles.values())<\/p>\n
# \u5206\u6790\u6240\u6709\u72b6\u6001\u7801
\n status_code_analysis = {}
\n for code in self.status_code_stats.keys():
\n analysis = self.analyze_by_status_code(code)
\n if analysis:
\n status_code_analysis[code] = analysis<\/p>\n
# \u5206\u6790\u6240\u6709\u7aef\u70b9
\n endpoint_analysis = {}
\n for endpoint in self.endpoint_stats.keys():
\n analysis = self.analyze_by_endpoint(endpoint)
\n if analysis:
\n endpoint_analysis[endpoint] = analysis<\/p>\n
# \u8bc6\u522b\u6700\u6162\u7684\u72b6\u6001\u7801
\n slowest_codes = sorted(
\n status_code_analysis.items(),
\n key=lambda x: x[1]['total_time_stats']['p95'],
\n reverse=True
\n )[:5]<\/p>\n
# \u8bc6\u522b\u6700\u6162\u7684\u7aef\u70b9
\n slowest_endpoints = sorted(
\n endpoint_analysis.items(),
\n key=lambda x: x[1]['total_time_stats']['p95'],
\n reverse=True
\n )[:5]<\/p>\n
return {
\n 'summary': {
\n 'total_requests': len(all_profiles),
\n 'unique_status_codes': len(status_code_analysis),
\n 'unique_endpoints': len(endpoint_analysis),
\n 'avg_response_time': statistics.mean([p.total_time_ms for p in all_profiles])
\n if all_profiles else 0
\n },
\n 'slowest_status_codes': [
\n {
\n 'status_code': code,
\n 'p95_response_time': analysis['total_time_stats']['p95'],
\n 'request_count': analysis['count']
\n }
\n for code, analysis in slowest_codes
\n ],
\n 'slowest_endpoints': [
\n {
\n 'endpoint': endpoint,
\n 'p95_response_time': analysis['total_time_stats']['p95'],
\n 'request_count': analysis['count']
\n }
\n for endpoint, analysis in slowest_endpoints
\n ],
\n 'common_bottlenecks': self._identify_common_bottlenecks(status_code_analysis),
\n 'performance_trends': self._analyze_overall_trends(all_profiles),
\n 'optimization_priority': self._calculate_optimization_priority(
\n slowest_codes, slowest_endpoints
\n )
\n }<\/p>\n
def _identify_common_bottlenecks(self, status_code_analysis: Dict) -> List[Dict]:
\n """\u8bc6\u522b\u5e38\u89c1\u74f6\u9888"""
\n bottleneck_counter = {}<\/p>\n
for code, analysis in status_code_analysis.items():
\n bottlenecks = analysis.get('bottlenecks', [])
\n for bottleneck in bottlenecks:
\n component = bottleneck['component']
\n if component not in bottleneck_counter:
\n bottleneck_counter[component] = 0
\n bottleneck_counter[component] += 1<\/p>\n
# \u8f6c\u6362\u4e3a\u767e\u5206\u6bd4
\n total_codes = len(status_code_analysis)
\n common_bottlenecks = [
\n {
\n 'component': component,
\n 'frequency': count,
\n 'percentage': (count \/ total_codes) * 100
\n }
\n for component, count in bottleneck_counter.items()
\n ]<\/p>\n
# \u6309\u9891\u7387\u6392\u5e8f
\n common_bottlenecks.sort(key=lambda x: x['frequency'], reverse=True)<\/p>\n
return common_bottlenecks<\/p>\n
def _analyze_overall_trends(self, profiles: List[RequestProfile]) -> Dict:
\n """\u5206\u6790\u603b\u4f53\u8d8b\u52bf"""
\n if len(profiles) < 20:
\n return {}<\/p>\n
# \u6309\u72b6\u6001\u7801\u5206\u7ec4\u8d8b\u52bf
\n trends_by_code = {}
\n for code in self.status_code_stats.keys():
\n code_profiles = self.status_code_stats[code]
\n if len(code_profiles) >= 10:
\n trends = self._analyze_trends(code_profiles)
\n if trends:
\n trends_by_code[code] = trends<\/p>\n
return {
\n 'trends_by_status_code': trends_by_code,
\n 'overall_trend': self._detect_trend(
\n [p.total_time_ms for p in sorted(profiles, key=lambda p: p.timings[0].start_time)]
\n )
\n }<\/p>\n
def _calculate_optimization_priority(self,
\n slowest_codes: List,
\n slowest_endpoints: List) -> List[Dict]:
\n """\u8ba1\u7b97\u4f18\u5316\u4f18\u5148\u7ea7"""
\n priorities = []<\/p>\n
# \u57fa\u4e8e\u54cd\u5e94\u65f6\u95f4\u548c\u8bf7\u6c42\u91cf\u8ba1\u7b97\u4f18\u5148\u7ea7\u5206\u6570
\n for code_info in slowest_codes:
\n code = code_info['status_code']
\n p95_time = code_info['p95_response_time']
\n count = code_info['request_count']<\/p>\n
# \u4f18\u5148\u7ea7\u5206\u6570 = \u54cd\u5e94\u65f6\u95f4 * log(\u8bf7\u6c42\u91cf)
\n priority_score = p95_time * (1 + (0.1 * count ** 0.5))<\/p>\n
priorities.append({
\n 'type': 'status_code',
\n 'id': code,
\n 'p95_response_time': p95_time,
\n 'request_count': count,
\n 'priority_score': priority_score,
\n 'priority_level': self._get_priority_level(priority_score)
\n })<\/p>\n
for endpoint_info in slowest_endpoints:
\n endpoint = endpoint_info['endpoint']
\n p95_time = endpoint_info['p95_response_time']
\n count = endpoint_info['request_count']<\/p>\n
priority_score = p95_time * (1 + (0.1 * count ** 0.5))<\/p>\n
priorities.append({
\n 'type': 'endpoint',
\n 'id': endpoint,
\n 'p95_response_time': p95_time,
\n 'request_count': count,
\n 'priority_score': priority_score,
\n 'priority_level': self._get_priority_level(priority_score)
\n })<\/p>\n
# \u6309\u4f18\u5148\u7ea7\u5206\u6570\u6392\u5e8f
\n priorities.sort(key=lambda x: x['priority_score'], reverse=True)<\/p>\n
return priorities<\/p>\n
def _get_priority_level(self, score: float) -> str:
\n """\u83b7\u53d6\u4f18\u5148\u7ea7\u7ea7\u522b"""
\n if score > 10000:
\n return 'CRITICAL'
\n elif score > 5000:
\n return 'HIGH'
\n elif score > 1000:
\n return 'MEDIUM'
\n else:
\n return 'LOW'<\/p>\n
# \u4f7f\u7528\u793a\u4f8b
\nanalyzer = PerformanceAnalyzer()<\/p>\n
# \u6a21\u62df\u4e00\u4e9b\u6027\u80fd\u6570\u636e
\nimport random
\nfrom datetime import datetime, timedelta<\/p>\n
for i in range(100):
\n request_id = f"req_{i}"
\n status_code = random.choice([200, 400, 404, 500])
\n total_time = random.uniform(50, 5000)<\/p>\n
# \u521b\u5efa\u65f6\u95f4\u6d4b\u91cf
\n timings = []
\n start = time.time()<\/p>\n
# \u6a21\u62df\u5404\u4e2a\u7ec4\u4ef6\u7684\u65f6\u95f4
\n components = [
\n (ResponseTimeComponent.NETWORK, 10, 100),
\n (ResponseTimeComponent.PROCESSING, 20, 1000),
\n (ResponseTimeComponent.DATABASE, 5, 500),
\n (ResponseTimeComponent.CACHE, 1, 50),
\n (ResponseTimeComponent.SERIALIZATION, 2, 100)
\n ]<\/p>\n
current_time = start
\n for component, min_time, max_time in components:
\n component_time = random.uniform(min_time, max_time) \/ 1000 # \u8f6c\u6362\u4e3a\u79d2
\n timings.append(TimingMeasurement(
\n component=component,
\n start_time=current_time,
\n end_time=current_time + component_time
\n ))
\n current_time += component_time<\/p>\n
profile = RequestProfile(
\n request_id=request_id,
\n status_code=status_code,
\n total_time_ms=total_time,
\n timings=timings,
\n endpoint=f"\/api\/resource\/{i % 5}",
\n method=random.choice(['GET', 'POST', 'PUT', 'DELETE'])
\n )<\/p>\n
analyzer.add_profile(profile)<\/p>\n
# \u5206\u6790200\u72b6\u6001\u7801\u7684\u6027\u80fd
\nanalysis_200 = analyzer.analyze_by_status_code(200)
\nprint(f"200\u54cd\u5e94\u5206\u6790: {analysis_200['count']} \u4e2a\u8bf7\u6c42")
\nprint(f"\u5e73\u5747\u54cd\u5e94\u65f6\u95f4: {analysis_200['total_time_stats']['mean']:.0f}ms")
\nprint(f"P95\u54cd\u5e94\u65f6\u95f4: {analysis_200['total_time_stats']['p95']:.0f}ms")<\/p>\n
# \u751f\u6210\u603b\u4f53\u62a5\u544a
\noverall_report = analyzer.generate_overall_report()
\nprint(f"\\\\n\u6700\u6162\u7684\u72b6\u6001\u7801:")
\nfor code_info in overall_report['slowest_status_codes']:
\n print(f" {code_info['status_code']}: P95={code_info['p95_response_time']:.0f}ms "
\n f"({code_info['request_count']} \u8bf7\u6c42)")<\/p>\n
print(f"\\\\n\u4f18\u5316\u4f18\u5148\u7ea7:")
\nfor priority in overall_report['optimization_priority'][:3]:
\n print(f" {priority['type']} {priority['id']}: {priority['priority_level']} "
\n f"(\u5206\u6570: {priority['priority_score']:.0f})")<\/p>\n
python<\/p>\n
# \u72b6\u6001\u7801\u54cd\u5e94\u7684\u7f13\u5b58\u4f18\u5316\u7b56\u7565
\nfrom typing import Dict, Any, Optional, Tuple
\nfrom dataclasses import dataclass
\nfrom datetime import datetime, timedelta
\nimport hashlib
\nimport json<\/p>\n
class CacheableResponse:
\n """\u53ef\u7f13\u5b58\u7684\u54cd\u5e94"""<\/p>\n
def __init__(self,
\n status_code: int,
\n headers: Dict[str, str],
\n body: Any,
\n cache_key: str,
\n ttl: timedelta = timedelta(minutes=5)):
\n self.status_code = status_code
\n self.headers = headers
\n self.body = body
\n self.cache_key = cache_key
\n self.created_at = datetime.now()
\n self.expires_at = self.created_at + ttl
\n self.hit_count = 0<\/p>\n
@property
\n def is_expired(self) -> bool:
\n return datetime.now() > self.expires_at<\/p>\n
@property
\n def age(self) -> timedelta:
\n return datetime.now() – self.created_at<\/p>\n
def record_hit(self):
\n """\u8bb0\u5f55\u547d\u4e2d"""
\n self.hit_count += 1<\/p>\n
def to_dict(self) -> Dict[str, Any]:
\n """\u8f6c\u6362\u4e3a\u5b57\u5178"""
\n return {
\n 'status_code': self.status_code,
\n 'headers': self.headers,
\n 'body': self.body,
\n 'cache_key': self.cache_key,
\n 'created_at': self.created_at.isoformat(),
\n 'expires_at': self.expires_at.isoformat(),
\n 'hit_count': self.hit_count,
\n 'age_seconds': self.age.total_seconds()
\n }<\/p>\n
class StatusCodeCacheStrategy:
\n """\u72b6\u6001\u7801\u7f13\u5b58\u7b56\u7565"""<\/p>\n
# \u4e0d\u540c\u72b6\u6001\u7801\u7684\u9ed8\u8ba4\u7f13\u5b58\u7b56\u7565
\n DEFAULT_CACHE_TTL = {
\n # \u6210\u529f\u54cd\u5e94
\n 200: timedelta(minutes=5), # \u9ed8\u8ba45\u5206\u949f
\n 201: timedelta(seconds=0), # POST\u521b\u5efa，\u901a\u5e38\u4e0d\u7f13\u5b58
\n 204: timedelta(minutes=1), # \u65e0\u5185\u5bb9，\u77ed\u65f6\u95f4\u7f13\u5b58<\/p>\n
# \u91cd\u5b9a\u5411
\n 301: timedelta(days=30), # \u6c38\u4e45\u91cd\u5b9a\u5411，\u957f\u65f6\u95f4\u7f13\u5b58
\n 302: timedelta(minutes=5), # \u4e34\u65f6\u91cd\u5b9a\u5411
\n 304: timedelta(minutes=5), # \u672a\u4fee\u6539，\u4f7f\u7528\u7f13\u5b58<\/p>\n
# \u5ba2\u6237\u7aef\u9519\u8bef
\n 400: timedelta(seconds=0), # \u9519\u8bef\u54cd\u5e94\u4e0d\u7f13\u5b58
\n 401: timedelta(seconds=0),
\n 403: timedelta(seconds=0),
\n 404: timedelta(minutes=1), # 404\u53ef\u77ed\u6682\u7f13\u5b58，\u907f\u514d\u91cd\u590d\u67e5\u8be2<\/p>\n
# \u670d\u52a1\u5668\u9519\u8bef
\n 500: timedelta(seconds=0), # \u9519\u8bef\u4e0d\u7f13\u5b58
\n 502: timedelta(seconds=0),
\n 503: timedelta(seconds=30), # \u670d\u52a1\u4e0d\u53ef\u7528，\u77ed\u6682\u7f13\u5b58\u907f\u514d\u96ea\u5d29
\n 504: timedelta(seconds=0),<\/p>\n
# \u81ea\u5b9a\u4e49\u72b6\u6001\u7801
\n 460: timedelta(minutes=1), # \u7f3a\u8d27\u72b6\u6001，\u77ed\u6682\u7f13\u5b58
\n 461: timedelta(minutes=1), # \u5e93\u5b58\u4e0d\u8db3
\n 520: timedelta(seconds=30), # \u7194\u65ad\u5668\u5f00\u542f，\u77ed\u6682\u7f13\u5b58
\n }<\/p>\n
# \u53ef\u7f13\u5b58\u7684\u72b6\u6001\u7801
\n CACHEABLE_STATUS_CODES = {
\n 200, 201, 202, 203, 204, 205, 206,
\n 300, 301, 302, 303, 304, 305, 306, 307,
\n 460, 461, 520 # \u81ea\u5b9a\u4e49\u72b6\u6001\u7801
\n }<\/p>\n
@classmethod
\n def is_cacheable(cls, status_code: int, method: str) -> bool:
\n """\u68c0\u67e5\u54cd\u5e94\u662f\u5426\u53ef\u7f13\u5b58"""
\n if method.upper() != 'GET':
\n return False<\/p>\n
return status_code in cls.CACHEABLE_STATUS_CODES<\/p>\n
@classmethod
\n def get_ttl_for_status(cls,
\n status_code: int,
\n endpoint: str,
\n headers: Dict[str, str]) -> timedelta:
\n """\u83b7\u53d6\u72b6\u6001\u7801\u7684TTL"""
\n # \u9996\u5148\u68c0\u67e5\u54cd\u5e94\u5934\u4e2d\u7684\u7f13\u5b58\u63a7\u5236
\n cache_control = headers.get('Cache-Control', '')
\n if 'max-age' in cache_control:
\n # \u89e3\u6790max-age
\n import re
\n match = re.search(r'max-age=(\\\\d+)', cache_control)
\n if match:
\n return timedelta(seconds=int(match.group(1)))<\/p>\n
# \u68c0\u67e5Expires\u5934\u90e8
\n expires = headers.get('Expires')
\n if expires:
\n try:
\n expires_dt = datetime.fromisoformat(expires.replace('Z', '+00:00'))
\n return expires_dt – datetime.now()
\n except:
\n pass<\/p>\n
# \u4f7f\u7528\u57fa\u4e8e\u72b6\u6001\u7801\u7684\u9ed8\u8ba4TTL
\n ttl = cls.DEFAULT_CACHE_TTL.get(status_code, timedelta(seconds=0))<\/p>\n
# \u6839\u636e\u7aef\u70b9\u8c03\u6574TTL
\n ttl = cls._adjust_ttl_by_endpoint(ttl, endpoint)<\/p>\n
return ttl<\/p>\n
@classmethod
\n def _adjust_ttl_by_endpoint(cls, ttl: timedelta, endpoint: str) -> timedelta:
\n """\u6839\u636e\u7aef\u70b9\u8c03\u6574TTL"""
\n # \u9759\u6001\u8d44\u6e90\u957f\u65f6\u95f4\u7f13\u5b58
\n static_extensions = ['.css', '.js', '.png', '.jpg', '.gif', '.ico', '.svg']
\n if any(endpoint.endswith(ext) for ext in static_extensions):
\n return timedelta(days=30)<\/p>\n
# API\u7aef\u70b9\u6839\u636e\u8def\u5f84\u8c03\u6574
\n if '\/api\/' in endpoint:
\n # \u516c\u5171API\u6570\u636e\u53ef\u7f13\u5b58\u66f4\u4e45
\n if '\/public\/' in endpoint or '\/catalog\/' in endpoint:
\n return max(ttl, timedelta(minutes=30))<\/p>\n
# \u7528\u6237\u76f8\u5173\u6570\u636e\u7f13\u5b58\u8f83\u77ed
\n if '\/user\/' in endpoint or '\/profile\/' in endpoint:
\n return min(ttl, timedelta(minutes=1))<\/p>\n
return ttl<\/p>\n
@classmethod
\n def generate_cache_key(cls,
\n method: str,
\n url: str,
\n headers: Dict[str, str],
\n body: Optional[Any] = None) -> str:
\n """\u751f\u6210\u7f13\u5b58\u952e"""
\n # \u57fa\u4e8e\u8bf7\u6c42\u7279\u5f81\u751f\u6210\u552f\u4e00\u952e
\n components = [
\n method.upper(),
\n url,
\n # \u89c4\u8303\u5316\u5934\u90e8（\u53ea\u5305\u542b\u5f71\u54cd\u54cd\u5e94\u7684\u5934\u90e8）
\n cls._normalize_headers(headers),
\n ]<\/p>\n
if body:
\n if isinstance(body, (dict, list)):
\n body_str = json.dumps(body, sort_keys=True)
\n else:
\n body_str = str(body)
\n components.append(body_str)<\/p>\n
# \u521b\u5efa\u54c8\u5e0c
\n key_string = '|'.join(str(c) for c in components)
\n return hashlib.sha256(key_string.encode()).hexdigest()<\/p>\n
@classmethod
\n def _normalize_headers(cls, headers: Dict[str, str]) -> str:
\n """\u89c4\u8303\u5316\u5934\u90e8"""
\n # \u53ea\u9009\u62e9\u5f71\u54cd\u54cd\u5e94\u7684\u5934\u90e8
\n relevant_headers = [
\n 'Authorization',
\n 'Accept',
\n 'Accept-Language',
\n 'Accept-Encoding',
\n 'User-Agent',
\n 'If-None-Match',
\n 'If-Modified-Since'
\n ]<\/p>\n
normalized = {}
\n for key in relevant_headers:
\n if key in headers:
\n normalized[key.lower()] = headers[key]<\/p>\n
# \u6392\u5e8f\u4ee5\u786e\u4fdd\u4e00\u81f4\u6027
\n sorted_items = sorted(normalized.items())
\n return json.dumps(sorted_items)<\/p>\n
class ResponseCache:
\n """\u54cd\u5e94\u7f13\u5b58"""<\/p>\n
def __init__(self, max_size: int = 10000):
\n self.max_size = max_size
\n self.cache: Dict[str, CacheableResponse] = {}
\n self.hit_count = 0
\n self.miss_count = 0
\n self.stats_by_status_code: Dict[int, Dict[str, int]] = {}<\/p>\n
def get(self, cache_key: str) -> Optional[CacheableResponse]:
\n """\u83b7\u53d6\u7f13\u5b58\u54cd\u5e94"""
\n if cache_key in self.cache:
\n cached = self.cache[cache_key]<\/p>\n
# \u68c0\u67e5\u662f\u5426\u8fc7\u671f
\n if cached.is_expired:
\n self._remove(cache_key)
\n self.miss_count += 1
\n return None<\/p>\n
# \u8bb0\u5f55\u547d\u4e2d
\n cached.record_hit()
\n self.hit_count += 1<\/p>\n
# \u66f4\u65b0\u7edf\u8ba1
\n self._update_stats(cached.status_code, 'hit')<\/p>\n
return cached<\/p>\n
self.miss_count += 1
\n return None<\/p>\n
def set(self, response: CacheableResponse):
\n """\u8bbe\u7f6e\u7f13\u5b58\u54cd\u5e94"""
\n # \u68c0\u67e5\u7f13\u5b58\u5927\u5c0f，\u5fc5\u8981\u65f6\u6e05\u7406
\n if len(self.cache) >= self.max_size:
\n self._evict()<\/p>\n
# \u6dfb\u52a0\u7f13\u5b58
\n self.cache[response.cache_key] = response<\/p>\n
# \u66f4\u65b0\u7edf\u8ba1
\n self._update_stats(response.status_code, 'store')<\/p>\n
def _evict(self):
\n """\u6e05\u7406\u7f13\u5b58"""
\n # \u4f7f\u7528LRU\u7b56\u7565
\n # \u9996\u5148\u79fb\u9664\u8fc7\u671f\u7684
\n expired_keys = [
\n key for key, response in self.cache.items()
\n if response.is_expired
\n ]<\/p>\n
for key in expired_keys:
\n self._remove(key)<\/p>\n
# \u5982\u679c\u4ecd\u7136\u9700\u8981\u7a7a\u95f4，\u79fb\u9664\u6700\u4e45\u672a\u4f7f\u7528\u7684
\n if len(self.cache) >= self.max_size:
\n # \u6309\u6700\u540e\u8bbf\u95ee\u65f6\u95f4\u6392\u5e8f（\u7b80\u5316\u5b9e\u73b0）
\n # \u5728\u5b9e\u9645\u4e2d，\u8fd9\u91cc\u9700\u8981\u7ef4\u62a4\u8bbf\u95ee\u987a\u5e8f
\n keys_to_remove = list(self.cache.keys())[:self.max_size \/\/ 10]
\n for key in keys_to_remove:
\n self._remove(key)<\/p>\n
def _remove(self, cache_key: str):
\n """\u79fb\u9664\u7f13\u5b58\u9879"""
\n if cache_key in self.cache:
\n response = self.cache[cache_key]
\n # \u66f4\u65b0\u7edf\u8ba1
\n self._update_stats(response.status_code, 'evict')
\n del self.cache[cache_key]<\/p>\n
def _update_stats(self, status_code: int, action: str):
\n """\u66f4\u65b0\u7edf\u8ba1"""
\n if status_code not in self.stats_by_status_code:
\n self.stats_by_status_code[status_code] = {
\n 'hit': 0,
\n 'store': 0,
\n 'evict': 0
\n }<\/p>\n
self.stats_by_status_code[status_code][action] += 1<\/p>\n
def get_stats(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u7f13\u5b58\u7edf\u8ba1"""
\n total_requests = self.hit_count + self.miss_count
\n hit_rate = self.hit_count \/ total_requests if total_requests > 0 else 0<\/p>\n
# \u8ba1\u7b97\u7f13\u5b58\u6548\u7387
\n efficiency_by_status = {}
\n for code, stats in self.stats_by_status_code.items():
\n stores = stats.get('store', 0)
\n hits = stats.get('hit', 0)
\n efficiency = hits \/ stores if stores > 0 else 0
\n efficiency_by_status[code] = efficiency<\/p>\n
return {
\n 'total_items': len(self.cache),
\n 'max_size': self.max_size,
\n 'hit_count': self.hit_count,
\n 'miss_count': self.miss_count,
\n 'hit_rate': hit_rate,
\n 'stats_by_status_code': self.stats_by_status_code,
\n 'efficiency_by_status': efficiency_by_status,
\n 'oldest_item': min(
\n (r.age for r in self.cache.values()),
\n default=timedelta(0)
\n ).total_seconds(),
\n 'newest_item': max(
\n (r.age for r in self.cache.values()),
\n default=timedelta(0)
\n ).total_seconds()
\n }<\/p>\n
def optimize_cache_strategy(self) -> Dict[str, Any]:
\n """\u4f18\u5316\u7f13\u5b58\u7b56\u7565"""
\n suggestions = []<\/p>\n
# \u5206\u6790\u7f13\u5b58\u6548\u7387
\n stats = self.get_stats()
\n efficiency_by_status = stats['efficiency_by_status']<\/p>\n
for code, efficiency in efficiency_by_status.items():
\n if efficiency < 0.1: # \u6548\u7387\u4f4e\u4e8e10%
\n suggestions.append({
\n 'status_code': code,
\n 'efficiency': efficiency,
\n 'suggestion': 'Consider reducing TTL or disabling caching'
\n })
\n elif efficiency > 0.8: # \u6548\u7387\u9ad8\u4e8e80%
\n suggestions.append({
\n 'status_code': code,
\n 'efficiency': efficiency,
\n 'suggestion': 'Consider increasing TTL'
\n })<\/p>\n
# \u5206\u6790\u7f13\u5b58\u547d\u4e2d\u7387
\n if stats['hit_rate'] < 0.3:
\n suggestions.append({
\n 'type': 'overall',
\n 'metric': 'hit_rate',
\n 'value': stats['hit_rate'],
\n 'suggestion': 'Overall cache hit rate is low. Consider adjusting cache keys or increasing cache size.'
\n })<\/p>\n
# \u5206\u6790\u7f13\u5b58\u5927\u5c0f
\n utilization = len(self.cache) \/ self.max_size
\n if utilization > 0.9:
\n suggestions.append({
\n 'type': 'size',
\n 'utilization': utilization,
\n 'suggestion': 'Cache is nearly full. Consider increasing max_size or improving eviction strategy.'
\n })<\/p>\n
return {
\n 'current_stats': stats,
\n 'suggestions': suggestions,
\n 'recommended_actions': self._generate_recommended_actions(suggestions)
\n }<\/p>\n
def _generate_recommended_actions(self, suggestions: List[Dict]) -> List[str]:
\n """\u751f\u6210\u63a8\u8350\u64cd\u4f5c"""
\n actions = []<\/p>\n
for suggestion in suggestions:
\n if suggestion.get('type') == 'overall':
\n actions.append("Review cache key generation strategy")
\n actions.append("Consider implementing multi-level caching")<\/p>\n
elif 'status_code' in suggestion:
\n code = suggestion['status_code']
\n if suggestion['efficiency'] < 0.1:
\n actions.append(f"Reduce TTL for status code {code}")
\n elif suggestion['efficiency'] > 0.8:
\n actions.append(f"Increase TTL for status code {code}")<\/p>\n
return list(set(actions)) # \u53bb\u91cd<\/p>\n
# \u7f13\u5b58\u4f18\u5316\u4e2d\u95f4\u4ef6
\nclass CacheOptimizationMiddleware:
\n """\u7f13\u5b58\u4f18\u5316\u4e2d\u95f4\u4ef6"""<\/p>\n
def __init__(self, cache: ResponseCache):
\n self.cache = cache
\n self.strategy = StatusCodeCacheStrategy()<\/p>\n
async def process_request(self, request: Dict) -> Optional[Dict]:
\n """\u5904\u7406\u8bf7\u6c42（\u7f13\u5b58\u67e5\u627e）"""
\n # \u751f\u6210\u7f13\u5b58\u952e
\n cache_key = self.strategy.generate_cache_key(
\n method=request['method'],
\n url=request['url'],
\n headers=request.get('headers', {}),
\n body=request.get('body')
\n )<\/p>\n
# \u68c0\u67e5\u7f13\u5b58
\n cached_response = self.cache.get(cache_key)
\n if cached_response:
\n # \u68c0\u67e5\u662f\u5426\u9700\u8981\u9a8c\u8bc1（ETag\/If-None-Match）
\n if self._should_validate(request, cached_response):
\n # \u6dfb\u52a0\u9a8c\u8bc1\u5934
\n request['headers']['If-None-Match'] = self._generate_etag(cached_response)<\/p>\n
# \u8fd4\u56de\u7f13\u5b58\u54cd\u5e94
\n return {
\n 'from_cache': True,
\n 'cache_key': cache_key,
\n 'response': cached_response
\n }<\/p>\n
return None<\/p>\n
async def process_response(self,
\n request: Dict,
\n response: Dict) -> Dict:
\n """\u5904\u7406\u54cd\u5e94（\u7f13\u5b58\u5b58\u50a8）"""
\n status_code = response['status_code']
\n method = request['method']<\/p>\n
# \u68c0\u67e5\u662f\u5426\u53ef\u7f13\u5b58
\n if not self.strategy.is_cacheable(status_code, method):
\n return response<\/p>\n
# \u751f\u6210\u7f13\u5b58\u952e
\n cache_key = self.strategy.generate_cache_key(
\n method=method,
\n url=request['url'],
\n headers=request.get('headers', {}),
\n body=request.get('body')
\n )<\/p>\n
# \u8ba1\u7b97TTL
\n ttl = self.strategy.get_ttl_for_status(
\n status_code=status_code,
\n endpoint=request['url'],
\n headers=response.get('headers', {})
\n )<\/p>\n
if ttl.total_seconds() > 0:
\n # \u521b\u5efa\u53ef\u7f13\u5b58\u54cd\u5e94
\n cacheable_response = CacheableResponse(
\n status_code=status_code,
\n headers=response.get('headers', {}),
\n body=response.get('body'),
\n cache_key=cache_key,
\n ttl=ttl
\n )<\/p>\n
# \u5b58\u50a8\u5230\u7f13\u5b58
\n self.cache.set(cacheable_response)<\/p>\n
# \u6dfb\u52a0\u7f13\u5b58\u5934\u5230\u54cd\u5e94
\n response['headers']['X-Cache'] = 'MISS'
\n response['headers']['X-Cache-Key'] = cache_key
\n response['headers']['X-Cache-TTL'] = str(int(ttl.total_seconds()))<\/p>\n
return response<\/p>\n
def _should_validate(self, request: Dict, cached_response: CacheableResponse) -> bool:
\n """\u68c0\u67e5\u662f\u5426\u9700\u8981\u9a8c\u8bc1\u7f13\u5b58"""
\n # \u5982\u679c\u6709ETag，\u901a\u5e38\u9700\u8981\u9a8c\u8bc1
\n if 'etag' in cached_response.headers:
\n return True<\/p>\n
# \u6839\u636e\u72b6\u6001\u7801\u51b3\u5b9a
\n status_code = cached_response.status_code
\n if status_code in [200, 203, 206]:
\n return True<\/p>\n
return False<\/p>\n
def _generate_etag(self, response: CacheableResponse) -> str:
\n """\u751f\u6210ETag"""
\n # \u57fa\u4e8e\u54cd\u5e94\u5185\u5bb9\u751f\u6210\u5f31ETag
\n content_hash = hashlib.md5(
\n json.dumps(response.body).encode()
\n ).hexdigest()<\/p>\n
return f'W\/"{content_hash}"'<\/p>\n
def get_optimization_report(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u4f18\u5316\u62a5\u544a"""
\n cache_stats = self.cache.get_stats()
\n optimization_suggestions = self.cache.optimize_cache_strategy()<\/p>\n
# \u8ba1\u7b97\u6027\u80fd\u6539\u8fdb
\n hit_rate = cache_stats['hit_rate']
\n estimated_savings = self._estimate_performance_savings(hit_rate)<\/p>\n
return {
\n 'cache_statistics': cache_stats,
\n 'optimization_suggestions': optimization_suggestions,
\n 'performance_impact': estimated_savings,
\n 'recommended_configuration': self._generate_recommended_configuration()
\n }<\/p>\n
def _estimate_performance_savings(self, hit_rate: float) -> Dict[str, float]:
\n """\u4f30\u8ba1\u6027\u80fd\u8282\u7701"""
\n # \u7b80\u5316\u4f30\u7b97
\n avg_cache_hit_time = 10 # ms – \u7f13\u5b58\u547d\u4e2d\u65f6\u95f4
\n avg_miss_time = 200 # ms – \u7f13\u5b58\u672a\u547d\u4e2d\u65f6\u95f4（\u9700\u8981\u5b8c\u6574\u5904\u7406）<\/p>\n
total_time_without_cache = avg_miss_time
\n total_time_with_cache = (hit_rate * avg_cache_hit_time +
\n (1 – hit_rate) * avg_miss_time)<\/p>\n
time_savings = total_time_without_cache – total_time_with_cache
\n percentage_savings = (time_savings \/ total_time_without_cache) * 100<\/p>\n
return {
\n 'avg_response_time_without_cache_ms': avg_miss_time,
\n 'avg_response_time_with_cache_ms': total_time_with_cache,
\n 'time_savings_ms': time_savings,
\n 'percentage_savings': percentage_savings,
\n 'estimated_requests_per_second_improvement':
\n (1 \/ total_time_with_cache * 1000) – (1 \/ avg_miss_time * 1000)
\n }<\/p>\n
def _generate_recommended_configuration(self) -> Dict[str, Any]:
\n """\u751f\u6210\u63a8\u8350\u914d\u7f6e"""
\n stats = self.cache.get_stats()<\/p>\n
# \u57fa\u4e8e\u547d\u4e2d\u7387\u8c03\u6574\u7f13\u5b58\u5927\u5c0f
\n hit_rate = stats['hit_rate']
\n if hit_rate > 0.8:
\n recommended_size = int(self.cache.max_size * 1.5)
\n elif hit_rate < 0.3:
\n recommended_size = int(self.cache.max_size * 0.7)
\n else:
\n recommended_size = self.cache.max_size<\/p>\n
# \u8c03\u6574TTL\u914d\u7f6e
\n ttl_adjustments = {}
\n for code, efficiency in stats['efficiency_by_status'].items():
\n if efficiency < 0.1:
\n current_ttl = StatusCodeCacheStrategy.DEFAULT_CACHE_TTL.get(
\n code, timedelta(seconds=0)
\n )
\n ttl_adjustments[code] = {
\n 'current': current_ttl.total_seconds(),
\n 'recommended': max(current_ttl.total_seconds() * 0.5, 1)
\n }
\n elif efficiency > 0.8:
\n current_ttl = StatusCodeCacheStrategy.DEFAULT_CACHE_TTL.get(
\n code, timedelta(seconds=0)
\n )
\n ttl_adjustments[code] = {
\n 'current': current_ttl.total_seconds(),
\n 'recommended': current_ttl.total_seconds() * 2
\n }<\/p>\n
return {
\n 'cache_size': {
\n 'current': self.cache.max_size,
\n 'recommended': recommended_size
\n },
\n 'ttl_adjustments': ttl_adjustments,
\n 'suggested_cacheable_codes': self._suggest_additional_cacheable_codes(stats)
\n }<\/p>\n
def _suggest_additional_cacheable_codes(self, stats: Dict) -> List[int]:
\n """\u5efa\u8bae\u989d\u5916\u7684\u53ef\u7f13\u5b58\u72b6\u6001\u7801"""
\n suggestions = []<\/p>\n
# \u5206\u6790\u54ea\u4e9b\u975e\u7f13\u5b58\u72b6\u6001\u7801\u9891\u7e41\u51fa\u73b0
\n # \u5728\u5b9e\u9645\u5b9e\u73b0\u4e2d，\u8fd9\u91cc\u9700\u8981\u8bbf\u95ee\u5386\u53f2\u6570\u636e<\/p>\n
# \u793a\u4f8b：\u5982\u679c404\u9891\u7e41\u51fa\u73b0\u4e14\u5185\u5bb9\u76f8\u540c，\u5efa\u8bae\u7f13\u5b58
\n if '404' in stats.get('frequent_errors', []):
\n suggestions.append(404)<\/p>\n
return suggestions<\/p>\n
# \u4f7f\u7528\u793a\u4f8b
\ncache = ResponseCache(max_size=1000)
\nmiddleware = CacheOptimizationMiddleware(cache)<\/p>\n
# \u6a21\u62df\u5904\u7406\u8bf7\u6c42
\nrequests = [
\n {
\n 'method': 'GET',
\n 'url': '\/api\/products\/123',
\n 'headers': {'Accept': 'application\/json'},
\n 'body': None
\n }
\n]<\/p>\n
for i in range(100):
\n request = {
\n 'method': 'GET',
\n 'url': f'\/api\/products\/{i % 10}',
\n 'headers': {'Accept': 'application\/json'},
\n 'body': None
\n }<\/p>\n
# \u68c0\u67e5\u7f13\u5b58
\n cached = middleware.process_request(request)<\/p>\n
if cached:
\n print(f"Cache hit for {request['url']}")
\n response = cached['response']
\n else:
\n # \u5904\u7406\u8bf7\u6c42
\n response = {
\n 'status_code': 200 if i % 20 != 0 else 404,
\n 'headers': {'Content-Type': 'application\/json'},
\n 'body': {'id': i % 10, 'name': f'Product {i % 10}'}
\n }<\/p>\n
# \u5904\u7406\u54cd\u5e94（\u53ef\u80fd\u7f13\u5b58）
\n response = middleware.process_response(request, response)<\/p>\n
# \u6bcf20\u4e2a\u8bf7\u6c42\u663e\u793a\u7edf\u8ba1
\n if i % 20 == 0 and i > 0:
\n stats = cache.get_stats()
\n print(f"\\\\nAfter {i} requests:")
\n print(f" Cache hits: {stats['hit_count']}")
\n print(f" Cache misses: {stats['miss_count']}")
\n print(f" Hit rate: {stats['hit_rate']:.2%}")<\/p>\n
# \u751f\u6210\u4f18\u5316\u62a5\u544a
\nreport = middleware.get_optimization_report()
\nprint(f"\\\\n=== Optimization Report ===")
\nprint(f"Cache hit rate: {report['cache_statistics']['hit_rate']:.2%}")
\nprint(f"Estimated time savings: {report['performance_impact']['percentage_savings']:.1f}%")
\nprint(f"\\\\nSuggestions:")
\nfor suggestion in report['optimization_suggestions']['suggestions'][:3]:
\n print(f" – {suggestion['suggestion']}")<\/p>\n
python<\/p>\n
# HTTP\/2\u5934\u90e8\u538b\u7f29\u4f18\u5316
\nfrom typing import Dict, List, Tuple
\nimport zlib
\nimport brotli<\/p>\n
class HeaderCompressionOptimizer:
\n """\u5934\u90e8\u538b\u7f29\u4f18\u5316\u5668"""<\/p>\n
def __init__(self):
\n self.static_table = self._load_static_table()
\n self.dynamic_table = []
\n self.dynamic_table_size = 4096 # \u9ed8\u8ba4\u5927\u5c0f
\n self.max_dynamic_table_size = 4096<\/p>\n
def _load_static_table(self) -> List[Tuple[str, str]]:
\n """\u52a0\u8f7d\u9759\u6001\u8868（HPACK）"""
\n # HPACK\u9759\u6001\u8868\u5b9a\u4e49
\n return [
\n (":authority", ""),
\n (":method", "GET"),
\n (":method", "POST"),
\n (":path", "\/"),
\n (":path", "\/index.html"),
\n (":scheme", "http"),
\n (":scheme", "https"),
\n (":status", "200"),
\n (":status", "204"),
\n (":status", "206"),
\n (":status", "304"),
\n (":status", "400"),
\n (":status", "404"),
\n (":status", "500"),
\n ("accept-charset", ""),
\n ("accept-encoding", "gzip, deflate"),
\n ("accept-language", ""),
\n ("accept-ranges", ""),
\n ("accept", ""),
\n ("access-control-allow-origin", ""),
\n ("age", ""),
\n ("allow", ""),
\n ("authorization", ""),
\n ("cache-control", ""),
\n ("content-disposition", ""),
\n ("content-encoding", ""),
\n ("content-language", ""),
\n ("content-length", ""),
\n ("content-location", ""),
\n ("content-range", ""),
\n ("content-type", ""),
\n ("cookie", ""),
\n ("date", ""),
\n ("etag", ""),
\n ("expect", ""),
\n ("expires", ""),
\n ("from", ""),
\n ("host", ""),
\n ("if-match", ""),
\n ("if-modified-since", ""),
\n ("if-none-match", ""),
\n ("if-range", ""),
\n ("if-unmodified-since", ""),
\n ("last-modified", ""),
\n ("link", ""),
\n ("location", ""),
\n ("max-forwards", ""),
\n ("proxy-authenticate", ""),
\n ("proxy-authorization", ""),
\n ("range", ""),
\n ("referer", ""),
\n ("refresh", ""),
\n ("retry-after", ""),
\n ("server", ""),
\n ("set-cookie", ""),
\n ("strict-transport-security", ""),
\n ("transfer-encoding", ""),
\n ("user-agent", ""),
\n ("vary", ""),
\n ("via", ""),
\n ("www-authenticate", "")
\n ]<\/p>\n
def compress_headers(self,
\n headers: Dict[str, str],
\n use_huffman: bool = True) -> bytes:
\n """\u538b\u7f29\u5934\u90e8"""
\n encoded_headers = []<\/p>\n
for name, value in headers.items():
\n # \u5c1d\u8bd5\u5728\u9759\u6001\u8868\u4e2d\u67e5\u627e
\n static_index = self._find_in_static_table(name, value)<\/p>\n
if static_index is not None:
\n # \u4f7f\u7528\u9759\u6001\u8868\u7d22\u5f15
\n encoded_headers.append(self._encode_indexed_header(static_index))
\n else:
\n # \u5c1d\u8bd5\u5728\u52a8\u6001\u8868\u4e2d\u67e5\u627e
\n dynamic_index = self._find_in_dynamic_table(name, value)<\/p>\n
if dynamic_index is not None:
\n # \u4f7f\u7528\u52a8\u6001\u8868\u7d22\u5f15
\n encoded_headers.append(
\n self._encode_indexed_header(
\n dynamic_index + len(self.static_table)
\n )
\n )
\n else:
\n # \u5b57\u9762\u7f16\u7801
\n encoded_headers.append(
\n self._encode_literal_header(name, value, use_huffman)
\n )<\/p>\n
# \u6dfb\u52a0\u5230\u52a8\u6001\u8868
\n self._add_to_dynamic_table(name, value)<\/p>\n
return b''.join(encoded_headers)<\/p>\n
def _find_in_static_table(self, name: str, value: str) -> Optional[int]:
\n """\u5728\u9759\u6001\u8868\u4e2d\u67e5\u627e"""
\n for i, (n, v) in enumerate(self.static_table):
\n if n == name and v == value:
\n return i + 1 # HPACK\u7d22\u5f15\u4ece1\u5f00\u59cb
\n return None<\/p>\n
def _find_in_dynamic_table(self, name: str, value: str) -> Optional[int]:
\n """\u5728\u52a8\u6001\u8868\u4e2d\u67e5\u627e"""
\n for i, (n, v) in enumerate(self.dynamic_table):
\n if n == name and v == value:
\n return i
\n return None<\/p>\n
def _encode_indexed_header(self, index: int) -> bytes:
\n """\u7f16\u7801\u7d22\u5f15\u5934\u90e8"""
\n # HPACK\u7d22\u5f15\u5934\u90e8\u683c\u5f0f：1xxxxxxx
\n return bytes([0x80 | (index & 0x7F)])<\/p>\n
def _encode_literal_header(self,
\n name: str,
\n value: str,
\n use_huffman: bool) -> bytes:
\n """\u7f16\u7801\u5b57\u9762\u5934\u90e8"""
\n # \u7b80\u5316\u7684\u5b9e\u73b0
\n # \u5728\u5b9e\u9645\u4e2d，\u8fd9\u91cc\u9700\u8981\u5b9e\u73b0\u5b8c\u6574\u7684HPACK\u7f16\u7801<\/p>\n
if use_huffman:
\n name_encoded = self._huffman_encode(name)
\n value_encoded = self._huffman_encode(value)
\n else:
\n name_encoded = name.encode('utf-8')
\n value_encoded = value.encode('utf-8')<\/p>\n
# \u7f16\u7801\u5934\u90e8
\n encoded = bytearray()<\/p>\n
# \u540d\u79f0\u957f\u5ea6
\n encoded.append(len(name_encoded))
\n encoded.extend(name_encoded)<\/p>\n
# \u503c\u957f\u5ea6
\n encoded.append(len(value_encoded))
\n encoded.extend(value_encoded)<\/p>\n
return bytes(encoded)<\/p>\n
def _huffman_encode(self, text: str) -> bytes:
\n """Huffman\u7f16\u7801（\u7b80\u5316）"""
\n # \u5728\u5b9e\u9645\u4e2d，\u8fd9\u91cc\u9700\u8981\u4f7f\u7528HPACK\u7684Huffman\u8868
\n return text.encode('utf-8')<\/p>\n
def _add_to_dynamic_table(self, name: str, value: str):
\n """\u6dfb\u52a0\u5230\u52a8\u6001\u8868"""
\n entry_size = len(name) + len(value) + 32 # \u4f30\u7b97\u5927\u5c0f<\/p>\n
# \u68c0\u67e5\u7a7a\u95f4
\n while (self._calculate_dynamic_table_size() + entry_size >
\n self.max_dynamic_table_size):
\n if self.dynamic_table:
\n self.dynamic_table.pop(0)
\n else:
\n break<\/p>\n
# \u6dfb\u52a0\u65b0\u6761\u76ee
\n self.dynamic_table.append((name, value))<\/p>\n
def _calculate_dynamic_table_size(self) -> int:
\n """\u8ba1\u7b97\u52a8\u6001\u8868\u5927\u5c0f"""
\n total = 0
\n for name, value in self.dynamic_table:
\n total += len(name) + len(value) + 32 # \u4f30\u7b97\u7684\u5f00\u9500
\n return total<\/p>\n
def analyze_header_patterns(self,
\n request_logs: List[Dict]) -> Dict[str, any]:
\n """\u5206\u6790\u5934\u90e8\u6a21\u5f0f"""
\n header_frequency = {}
\n header_value_patterns = {}<\/p>\n
for log in request_logs:
\n headers = log.get('headers', {})<\/p>\n
for name, value in headers.items():
\n # \u7edf\u8ba1\u5934\u90e8\u9891\u7387
\n if name not in header_frequency:
\n header_frequency[name] = 0
\n header_frequency[name] += 1<\/p>\n
# \u5206\u6790\u503c\u6a21\u5f0f
\n if name not in header_value_patterns:
\n header_value_patterns[name] = {}<\/p>\n
if value not in header_value_patterns[name]:
\n header_value_patterns[name][value] = 0
\n header_value_patterns[name][value] += 1<\/p>\n
# \u627e\u51fa\u6700\u5e38\u7528\u7684\u5934\u90e8
\n common_headers = sorted(
\n header_frequency.items(),
\n key=lambda x: x[1],
\n reverse=True
\n )[:20]<\/p>\n
# \u5206\u6790\u503c\u5206\u5e03
\n value_distribution = {}
\n for name, patterns in header_value_patterns.items():
\n total = sum(patterns.values())
\n if total > 10: # \u8db3\u591f\u6837\u672c
\n most_common = max(patterns.items(), key=lambda x: x[1])
\n value_distribution[name] = {
\n 'total_values': len(patterns),
\n 'most_common_value': most_common[0],
\n 'most_common_percentage': most_common[1] \/ total,
\n 'entropy': self._calculate_entropy(patterns.values())
\n }<\/p>\n
return {
\n 'common_headers': common_headers,
\n 'value_distribution': value_distribution,
\n 'compression_opportunities': self._identify_compression_opportunities(
\n common_headers, value_distribution
\n )
\n }<\/p>\n
def _calculate_entropy(self, frequencies: List[int]) -> float:
\n """\u8ba1\u7b97\u71b5"""
\n import math<\/p>\n
total = sum(frequencies)
\n if total == 0:
\n return 0<\/p>\n
entropy = 0
\n for freq in frequencies:
\n probability = freq \/ total
\n if probability > 0:
\n entropy -= probability * math.log2(probability)<\/p>\n
return entropy<\/p>\n
def _identify_compression_opportunities(self,
\n common_headers: List[Tuple[str, int]],
\n value_distribution: Dict) -> List[Dict]:
\n """\u8bc6\u522b\u538b\u7f29\u673a\u4f1a"""
\n opportunities = []<\/p>\n
for name, frequency in common_headers:
\n if name in value_distribution:
\n dist = value_distribution[name]<\/p>\n
# \u9ad8\u9891\u7387\u4e14\u4f4e\u71b5\u7684\u503c\u662f\u597d\u7684\u538b\u7f29\u5019\u9009
\n if (frequency > 100 and
\n dist['most_common_percentage'] > 0.8 and
\n dist['entropy'] < 1.0):<\/p>\n
opportunities.append({
\n 'header': name,
\n 'frequency': frequency,
\n 'most_common_value': dist['most_common_value'],
\n 'value_consistency': dist['most_common_percentage'],
\n 'entropy': dist['entropy'],
\n 'compression_potential': self._estimate_compression_potential(
\n name, dist['most_common_value'], frequency
\n )
\n })<\/p>\n
return sorted(opportunities,
\n key=lambda x: x['compression_potential'],
\n reverse=True)<\/p>\n
def _estimate_compression_potential(self,
\n name: str,
\n value: str,
\n frequency: int) -> float:
\n """\u4f30\u8ba1\u538b\u7f29\u6f5c\u529b"""
\n # \u539f\u59cb\u5927\u5c0f
\n original_size = (len(name) + len(value)) * frequency<\/p>\n
# \u538b\u7f29\u540e\u5927\u5c0f（\u4f30\u8ba1）
\n # \u4f7f\u7528\u9759\u6001\u8868：1\u5b57\u8282
\n # \u4f7f\u7528\u52a8\u6001\u8868：~5\u5b57\u8282
\n # \u5b57\u9762\u7f16\u7801：~10+\u5b57\u8282<\/p>\n
if self._find_in_static_table(name, value):
\n compressed_size = frequency * 1
\n elif frequency > 10:
\n compressed_size = frequency * 5 # \u52a8\u6001\u8868
\n else:
\n compressed_size = frequency * (len(name) + len(value) + 2)<\/p>\n
savings = original_size – compressed_size
\n return savings<\/p>\n
def optimize_compression_strategy(self,
\n analysis: Dict) -> Dict[str, any]:
\n """\u4f18\u5316\u538b\u7f29\u7b56\u7565"""
\n recommendations = []<\/p>\n
# \u57fa\u4e8e\u5e38\u89c1\u5934\u90e8\u8c03\u6574\u9759\u6001\u8868
\n common_headers = analysis['common_headers'][:10]
\n static_table_candidates = [
\n {'header': name, 'frequency': freq}
\n for name, freq in common_headers
\n if freq > 1000 # \u975e\u5e38\u5e38\u89c1\u7684\u5934\u90e8
\n ]<\/p>\n
if static_table_candidates:
\n recommendations.append({
\n 'type': 'static_table_extension',
\n 'candidates': static_table_candidates,
\n 'estimated_savings': sum(
\n c['frequency'] * 10 # \u6bcf\u4e2a\u5934\u90e8\u8282\u7701\u7ea610\u5b57\u8282
\n for c in static_table_candidates
\n )
\n })<\/p>\n
# \u8c03\u6574\u52a8\u6001\u8868\u5927\u5c0f
\n current_size = self.max_dynamic_table_size
\n suggested_size = self._calculate_optimal_table_size(analysis)<\/p>\n
if suggested_size != current_size:
\n recommendations.append({
\n 'type': 'dynamic_table_size',
\n 'current': current_size,
\n 'suggested': suggested_size,
\n 'reason': 'Based on header pattern analysis'
\n })<\/p>\n
# Huffman\u7f16\u7801\u4f18\u5316
\n low_entropy_headers = [
\n opp for opp in analysis['compression_opportunities']
\n if opp['entropy'] < 0.5
\n ]<\/p>\n
if low_entropy_headers:
\n recommendations.append({
\n 'type': 'huffman_optimization',
\n 'headers': low_entropy_headers[:5],
\n 'suggestion': 'These headers have low entropy values, '
\n 'Huffman coding would be highly effective'
\n })<\/p>\n
return {
\n 'current_configuration': {
\n 'static_table_size': len(self.static_table),
\n 'dynamic_table_size': self.max_dynamic_table_size,
\n 'use_huffman': True
\n },
\n 'recommendations': recommendations,
\n 'estimated_overall_savings': sum(
\n r.get('estimated_savings', 0) for r in recommendations
\n )
\n }<\/p>\n
def _calculate_optimal_table_size(self, analysis: Dict) -> int:
\n """\u8ba1\u7b97\u6700\u4f18\u8868\u5927\u5c0f"""
\n # \u57fa\u4e8e\u5934\u90e8\u9891\u7387\u548c\u503c\u5206\u5e03
\n total_headers = sum(freq for _, freq in analysis['common_headers'])<\/p>\n
if total_headers > 10000:
\n return 8192 # 8KB
\n elif total_headers > 1000:
\n return 4096 # 4KB
\n else:
\n return 2048 # 2KB<\/p>\n
# HTTP\/3 QPACK\u4f18\u5316
\nclass QPACKOptimizer:
\n """QPACK\u4f18\u5316\u5668（HTTP\/3）"""<\/p>\n
def __init__(self):
\n self.static_table = self._load_qpack_static_table()
\n self.dynamic_table = []
\n self.max_table_capacity = 0
\n self.blocked_streams_allowed = True<\/p>\n
def _load_qpack_static_table(self) -> List[Tuple[str, str]]:
\n """\u52a0\u8f7dQPACK\u9759\u6001\u8868"""
\n # QPACK\u9759\u6001\u8868\u4e0eHPACK\u7c7b\u4f3c，\u4f46\u6709\u4e9b\u6269\u5c55
\n table = []<\/p>\n
# \u72b6\u6001\u7801\u6761\u76ee
\n for code in range(100, 600):
\n table.append((":status", str(code)))<\/p>\n
# \u5e38\u89c1\u5934\u90e8
\n common_headers = [
\n ("content-type", "application\/json"),
\n ("content-type", "text\/html"),
\n ("content-type", "application\/javascript"),
\n ("cache-control", "max-age=3600"),
\n ("cache-control", "no-cache"),
\n ("authorization", "Bearer"),
\n ("user-agent", "Mozilla\/5.0"),
\n ("accept-encoding", "gzip, br"),
\n ]<\/p>\n
table.extend(common_headers)
\n return table<\/p>\n
def optimize_for_stream(self,
\n stream_id: int,
\n headers: Dict[str, str]) -> Dict[str, any]:
\n """\u4e3a\u7279\u5b9a\u6d41\u4f18\u5316"""
\n # QPACK\u652f\u6301\u6bcf\u4e2a\u6d41\u72ec\u7acb\u7684\u52a8\u6001\u8868\u5f15\u7528
\n stream_optimization = {
\n 'stream_id': stream_id,
\n 'suggested_encodings': [],
\n 'table_references': [],
\n 'estimated_size': 0
\n }<\/p>\n
for name, value in headers.items():
\n encoding = self._suggest_encoding(name, value, stream_id)
\n stream_optimization['suggested_encodings'].append(encoding)
\n stream_optimization['estimated_size'] += encoding['estimated_size']<\/p>\n
return stream_optimization<\/p>\n
def _suggest_encoding(self,
\n name: str,
\n value: str,
\n stream_id: int) -> Dict[str, any]:
\n """\u5efa\u8bae\u7f16\u7801\u7b56\u7565"""
\n # \u68c0\u67e5\u9759\u6001\u8868
\n static_index = self._find_in_qpack_static_table(name, value)
\n if static_index is not None:
\n return {
\n 'type': 'static_reference',
\n 'index': static_index,
\n 'estimated_size': 1, # 1\u5b57\u8282
\n 'efficiency': 'high'
\n }<\/p>\n
# \u68c0\u67e5\u52a8\u6001\u8868
\n dynamic_index = self._find_in_dynamic_table(name, value)
\n if dynamic_index is not None:
\n return {
\n 'type': 'dynamic_reference',
\n 'index': dynamic_index,
\n 'estimated_size': 2, # \u7ea62\u5b57\u8282
\n 'efficiency': 'medium'
\n }<\/p>\n
# \u68c0\u67e5\u540d\u79f0\u5f15\u7528
\n name_index = self._find_name_in_tables(name)
\n if name_index is not None:
\n return {
\n 'type': 'name_reference',
\n 'index': name_index,
\n 'value': value,
\n 'estimated_size': len(value) + 2, # \u503c\u957f\u5ea6+\u5f00\u9500
\n 'efficiency': 'low'
\n }<\/p>\n
# \u5b57\u9762\u7f16\u7801
\n return {
\n 'type': 'literal',
\n 'name': name,
\n 'value': value,
\n 'estimated_size': len(name) + len(value) + 4, # \u540d\u79f0+\u503c+\u5f00\u9500
\n 'efficiency': 'very_low'
\n }<\/p>\n
def _find_in_qpack_static_table(self, name: str, value: str) -> Optional[int]:
\n """\u5728QPACK\u9759\u6001\u8868\u4e2d\u67e5\u627e"""
\n for i, (n, v) in enumerate(self.static_table):
\n if n == name and v == value:
\n return i
\n return None<\/p>\n
def _find_in_dynamic_table(self, name: str, value: str) -> Optional[int]:
\n """\u5728\u52a8\u6001\u8868\u4e2d\u67e5\u627e"""
\n for i, (n, v) in enumerate(self.dynamic_table):
\n if n == name and v == value:
\n return i
\n return None<\/p>\n
def _find_name_in_tables(self, name: str) -> Optional[int]:
\n """\u5728\u8868\u4e2d\u67e5\u627e\u540d\u79f0"""
\n # \u68c0\u67e5\u9759\u6001\u8868
\n for i, (n, _) in enumerate(self.static_table):
\n if n == name:
\n return i<\/p>\n
# \u68c0\u67e5\u52a8\u6001\u8868
\n for i, (n, _) in enumerate(self.dynamic_table):
\n if n == name:
\n return i + len(self.static_table)<\/p>\n
return None<\/p>\n
# \u7efc\u5408\u4f18\u5316\u5668
\nclass HTTPHeaderOptimizer:
\n """HTTP\u5934\u90e8\u7efc\u5408\u4f18\u5316\u5668"""<\/p>\n
def __init__(self):
\n self.hpack_optimizer = HeaderCompressionOptimizer()
\n self.qpack_optimizer = QPACKOptimizer()
\n self.protocol_analytics = {}<\/p>\n
def analyze_protocol_performance(self,
\n http_version: str,
\n request_logs: List[Dict]) -> Dict[str, any]:
\n """\u5206\u6790\u534f\u8bae\u6027\u80fd"""
\n # \u5206\u6790\u5934\u90e8\u538b\u7f29\u6548\u7387
\n header_analysis = self.hpack_optimizer.analyze_header_patterns(request_logs)<\/p>\n
# \u8ba1\u7b97\u7406\u8bba\u538b\u7f29\u6bd4
\n total_original_size = 0
\n total_compressed_size = 0<\/p>\n
for log in request_logs:
\n headers = log.get('headers', {})<\/p>\n
# \u539f\u59cb\u5927\u5c0f
\n for name, value in headers.items():
\n total_original_size += len(name) + len(value)<\/p>\n
# \u538b\u7f29\u5927\u5c0f（\u4f30\u7b97）
\n compressed = self.hpack_optimizer.compress_headers(headers)
\n total_compressed_size += len(compressed)<\/p>\n
compression_ratio = total_compressed_size \/ total_original_size if total_original_size > 0 else 0<\/p>\n
# \u534f\u8bae\u7279\u5b9a\u5efa\u8bae
\n protocol_suggestions = self._generate_protocol_suggestions(
\n http_version, header_analysis, compression_ratio
\n )<\/p>\n
return {
\n 'http_version': http_version,
\n 'compression_ratio': compression_ratio,
\n 'estimated_bandwidth_savings': total_original_size – total_compressed_size,
\n 'header_analysis': header_analysis,
\n 'protocol_suggestions': protocol_suggestions,
\n 'migration_recommendation': self._evaluate_protocol_migration(
\n http_version, compression_ratio
\n )
\n }<\/p>\n
def _generate_protocol_suggestions(self,
\n http_version: str,
\n header_analysis: Dict,
\n compression_ratio: float) -> List[Dict]:
\n """\u751f\u6210\u534f\u8bae\u7279\u5b9a\u5efa\u8bae"""
\n suggestions = []<\/p>\n
if http_version == 'HTTP\/1.1':
\n if compression_ratio > 0.7: # \u538b\u7f29\u7387\u4f4e
\n suggestions.append({
\n 'type': 'protocol_upgrade',
\n 'suggestion': 'Consider upgrading to HTTP\/2 for better header compression',
\n 'priority': 'high'
\n })<\/p>\n
# \u5934\u90e8\u5408\u5e76\u5efa\u8bae
\n suggestions.append({
\n 'type': 'header_consolidation',
\n 'suggestion': 'Consolidate multiple Set-Cookie headers into one',
\n 'priority': 'medium'
\n })<\/p>\n
elif http_version == 'HTTP\/2':
\n if compression_ratio > 0.5: # \u4ecd\u6709\u6539\u8fdb\u7a7a\u95f4
\n suggestions.append({
\n 'type': 'hpack_optimization',
\n 'suggestion': 'Optimize HPACK dynamic table size based on header patterns',
\n 'priority': 'medium'
\n })<\/p>\n
# \u670d\u52a1\u5668\u63a8\u9001\u4f18\u5316
\n suggestions.append({
\n 'type': 'server_push_optimization',
\n 'suggestion': 'Use server push for common resources referenced in responses',
\n 'priority': 'low'
\n })<\/p>\n
elif http_version == 'HTTP\/3':
\n if compression_ratio > 0.4:
\n suggestions.append({
\n 'type': 'qpack_optimization',
\n 'suggestion': 'Optimize QPACK configuration for better stream independence',
\n 'priority': 'medium'
\n })<\/p>\n
# \u901a\u7528\u4f18\u5316
\n opportunities = header_analysis.get('compression_opportunities', [])
\n if opportunities:
\n suggestions.append({
\n 'type': 'header_value_optimization',
\n 'suggestion': 'Standardize header values to improve compression',
\n 'examples': [opp['header'] for opp in opportunities[:3]],
\n 'priority': 'high'
\n })<\/p>\n
return suggestions<\/p>\n
def _evaluate_protocol_migration(self,
\n current_version: str,
\n compression_ratio: float) -> Optional[Dict]:
\n """\u8bc4\u4f30\u534f\u8bae\u8fc1\u79fb\u5efa\u8bae"""
\n if current_version == 'HTTP\/1.1' and compression_ratio > 0.6:
\n return {
\n 'recommended_protocol': 'HTTP\/2',
\n 'expected_improvement': f"{(1 – 0.3\/compression_ratio)*100:.1f}%", # \u4f30\u7b97
\n 'complexity': 'medium',
\n 'prerequisites': ['TLS 1.2+', 'Modern client support']
\n }<\/p>\n
elif current_version == 'HTTP\/2' and compression_ratio > 0.4:
\n return {
\n 'recommended_protocol': 'HTTP\/3',
\n 'expected_improvement': f"{(1 – 0.2\/compression_ratio)*100:.1f}%", # \u4f30\u7b97
\n 'complexity': 'high',
\n 'prerequisites': ['QUIC support', 'Updated infrastructure']
\n }<\/p>\n
return None<\/p>\n
def generate_optimization_plan(self,
\n analysis_results: Dict) -> Dict[str, any]:
\n """\u751f\u6210\u4f18\u5316\u8ba1\u5212"""
\n plan = {
\n 'immediate_actions': [],
\n 'short_term_actions': [],
\n 'long_term_actions': [],
\n 'expected_benefits': {},
\n 'implementation_complexity': {}
\n }<\/p>\n
# \u5206\u7c7b\u5efa\u8bae
\n for suggestion in analysis_results.get('protocol_suggestions', []):
\n priority = suggestion.get('priority', 'medium')<\/p>\n
if priority == 'high':
\n plan['immediate_actions'].append(suggestion)
\n elif priority == 'medium':
\n plan['short_term_actions'].append(suggestion)
\n else:
\n plan['long_term_actions'].append(suggestion)<\/p>\n
# \u8fc1\u79fb\u5efa\u8bae
\n migration = analysis_results.get('migration_recommendation')
\n if migration:
\n plan['long_term_actions'].append({
\n 'type': 'protocol_migration',
\n 'suggestion': f"Migrate to {migration['recommended_protocol']}",
\n 'details': migration
\n })<\/p>\n
# \u8ba1\u7b97\u9884\u671f\u6536\u76ca
\n plan['expected_benefits'] = self._calculate_expected_benefits(plan, analysis_results)<\/p>\n
# \u8bc4\u4f30\u5b9e\u73b0\u590d\u6742\u5ea6
\n plan['implementation_complexity'] = self._assess_complexity(plan)<\/p>\n
return plan<\/p>\n
def _calculate_expected_benefits(self,
\n plan: Dict,
\n analysis: Dict) -> Dict[str, float]:
\n """\u8ba1\u7b97\u9884\u671f\u6536\u76ca"""
\n benefits = {
\n 'bandwidth_reduction_percentage': 0,
\n 'latency_reduction_ms': 0,
\n 'throughput_increase_percentage': 0
\n }<\/p>\n
# \u57fa\u4e8e\u538b\u7f29\u6bd4\u4f30\u7b97
\n compression_ratio = analysis.get('compression_ratio', 1)<\/p>\n
# \u5934\u90e8\u538b\u7f29\u6536\u76ca
\n if compression_ratio < 1:
\n bandwidth_reduction = (1 – compression_ratio) * 100<\/p>\n
# \u5e94\u7528\u4f18\u5316\u4e58\u6570
\n optimization_multiplier = 1.0
\n if plan['immediate_actions']:
\n optimization_multiplier *= 0.8 # 20%\u989d\u5916\u6539\u8fdb
\n if plan['short_term_actions']:
\n optimization_multiplier *= 0.9 # 10%\u989d\u5916\u6539\u8fdb<\/p>\n
benefits['bandwidth_reduction_percentage'] = bandwidth_reduction * optimization_multiplier<\/p>\n
# \u5ef6\u8fdf\u6536\u76ca（\u4f30\u7b97）
\n avg_header_size = 500 # \u5b57\u8282
\n network_speed = 10 # Mbps
\n latency_reduction = (avg_header_size * (1 – compression_ratio) * 8) \/ (network_speed * 1000)
\n benefits['latency_reduction_ms'] = latency_reduction * 1000 # \u8f6c\u6362\u4e3a\u6beb\u79d2<\/p>\n
# \u534f\u8bae\u8fc1\u79fb\u6536\u76ca
\n migration = analysis.get('migration_recommendation')
\n if migration:
\n expected_improvement = migration.get('expected_improvement', '0%')
\n improvement = float(expected_improvement.strip('%'))
\n benefits['throughput_increase_percentage'] += improvement<\/p>\n
return benefits<\/p>\n
def _assess_complexity(self, plan: Dict) -> Dict[str, str]:
\n """\u8bc4\u4f30\u5b9e\u73b0\u590d\u6742\u5ea6"""
\n complexity = {
\n 'overall': 'low',
\n 'immediate_actions': 'low',
\n 'short_term_actions': 'medium',
\n 'long_term_actions': 'high'
\n }<\/p>\n
# \u57fa\u4e8e\u884c\u52a8\u7c7b\u578b\u8c03\u6574
\n for category in ['immediate_actions', 'short_term_actions', 'long_term_actions']:
\n actions = plan[category]
\n if actions:
\n # \u68c0\u67e5\u662f\u5426\u6709\u590d\u6742\u884c\u52a8
\n complex_actions = [a for a in actions if a.get('type') in
\n ['protocol_migration', 'protocol_upgrade']]
\n if complex_actions:
\n complexity[category] = 'high'<\/p>\n
# \u603b\u4f53\u590d\u6742\u5ea6
\n if complexity['long_term_actions'] == 'high':
\n complexity['overall'] = 'medium'
\n if complexity['short_term_actions'] == 'high':
\n complexity['overall'] = 'high'<\/p>\n
return complexity<\/p>\n
# \u4f7f\u7528\u793a\u4f8b
\noptimizer = HTTPHeaderOptimizer()<\/p>\n
# \u6a21\u62df\u8bf7\u6c42\u65e5\u5fd7
\nrequest_logs = []
\nfor i in range(100):
\n request_logs.append({
\n 'headers': {
\n 'user-agent': 'Mozilla\/5.0',
\n 'accept': 'application\/json',
\n 'content-type': 'application\/json',
\n 'authorization': 'Bearer token123',
\n 'cache-control': 'no-cache'
\n }
\n })<\/p>\n
# \u5206\u6790\u6027\u80fd
\nanalysis = optimizer.analyze_protocol_performance('HTTP\/1.1', request_logs)<\/p>\n
print(f"Compression ratio: {analysis['compression_ratio']:.2%}")
\nprint(f"Bandwidth savings: {analysis['estimated_bandwidth_savings']:,} bytes")<\/p>\n
# \u751f\u6210\u4f18\u5316\u8ba1\u5212
\nplan = optimizer.generate_optimization_plan(analysis)<\/p>\n
print("\\\\n=== Optimization Plan ===")
\nprint(f"Immediate actions ({len(plan['immediate_actions'])}):")
\nfor action in plan['immediate_actions']:
\n print(f" – {action['suggestion']}")<\/p>\n
print(f"\\\\nExpected benefits:")
\nfor benefit, value in plan['expected_benefits'].items():
\n print(f" {benefit}: {value}")<\/p>\n
python<\/p>\n
# CDN\u72b6\u6001\u7801\u4f18\u5316\u7b56\u7565
\nfrom typing import Dict, List, Optional, Set
\nfrom dataclasses import dataclass
\nfrom datetime import datetime, timedelta
\nimport hashlib<\/p>\n
@dataclass
\nclass CDNEdgeConfig:
\n """CDN\u8fb9\u7f18\u8282\u70b9\u914d\u7f6e"""
\n location: str
\n cache_ttl: Dict[int, int] # \u72b6\u6001\u7801 -> TTL（\u79d2）
\n stale_while_revalidate: bool
\n stale_if_error: bool
\n compression_enabled: bool
\n brotli_enabled: bool<\/p>\n
@classmethod
\n def default_config(cls) -> 'CDNEdgeConfig':
\n """\u9ed8\u8ba4\u914d\u7f6e"""
\n return cls(
\n location='global',
\n cache_ttl={
\n 200: 3600, # 1\u5c0f\u65f6
\n 201: 0, # \u4e0d\u7f13\u5b58
\n 204: 300, # 5\u5206\u949f
\n 301: 2592000, # 30\u5929
\n 302: 300, # 5\u5206\u949f
\n 304: 300, # 5\u5206\u949f
\n 404: 60, # 1\u5206\u949f
\n 500: 0, # \u4e0d\u7f13\u5b58
\n 502: 0,
\n 503: 30, # 30\u79d2（\u670d\u52a1\u4e0d\u53ef\u7528）
\n 504: 0,
\n 460: 60, # 1\u5206\u949f（\u7f3a\u8d27）
\n 520: 30 # 30\u79d2（\u7194\u65ad）
\n },
\n stale_while_revalidate=True,
\n stale_if_error=True,
\n compression_enabled=True,
\n brotli_enabled=True
\n )<\/p>\n
@dataclass
\nclass CDNCacheKey:
\n """CDN\u7f13\u5b58\u952e"""
\n url: str
\n status_code: int
\n vary_headers: Dict[str, str]
\n geo_location: Optional[str]
\n device_type: Optional[str]<\/p>\n
def generate_key(self) -> str:
\n """\u751f\u6210\u7f13\u5b58\u952e"""
\n components = [
\n self.url,
\n str(self.status_code),
\n # \u89c4\u8303\u5316vary\u5934\u90e8
\n self._normalize_vary_headers(),
\n self.geo_location or '',
\n self.device_type or ''
\n ]<\/p>\n
key_string = '|'.join(components)
\n return hashlib.md5(key_string.encode()).hexdigest()<\/p>\n
def _normalize_vary_headers(self) -> str:
\n """\u89c4\u8303\u5316Vary\u5934\u90e8"""
\n sorted_items = sorted(self.vary_headers.items())
\n return ';'.join(f"{k}:{v}" for k, v in sorted_items)<\/p>\n
class CDNOptimizer:
\n """CDN\u4f18\u5316\u5668"""<\/p>\n
def __init__(self, edge_config: CDNEdgeConfig):
\n self.edge_config = edge_config
\n self.cache_analytics = {}
\n self.hit_rate_by_status: Dict[int, Dict[str, float]] = {}<\/p>\n
def generate_cache_headers(self,
\n status_code: int,
\n content_type: str,
\n is_public: bool = True) -> Dict[str, str]:
\n """\u751f\u6210\u7f13\u5b58\u5934\u90e8"""
\n headers = {}<\/p>\n
# Cache-Control
\n cache_control = []<\/p>\n
ttl = self.edge_config.cache_ttl.get(status_code, 0)<\/p>\n
if ttl > 0:
\n if is_public:
\n cache_control.append(f"public")
\n cache_control.append(f"max-age={ttl}")
\n else:
\n cache_control.append(f"private")
\n cache_control.append(f"max-age={ttl}")<\/p>\n
# Stale-while-revalidate
\n if self.edge_config.stale_while_revalidate:
\n cache_control.append(f"stale-while-revalidate={ttl\/\/2}")<\/p>\n
# Stale-if-error
\n if self.edge_config.stale_if_error:
\n cache_control.append(f"stale-if-error={ttl*2}")
\n else:
\n cache_control.append("no-cache")
\n cache_control.append("must-revalidate")<\/p>\n
headers['Cache-Control'] = ', '.join(cache_control)<\/p>\n
# Expires（\u5411\u540e\u517c\u5bb9）
\n if ttl > 0:
\n expires = datetime.utcnow() + timedelta(seconds=ttl)
\n headers['Expires'] = expires.strftime('%a, %d %b %Y %H:%M:%S GMT')<\/p>\n
# Vary\u5934\u90e8
\n vary_headers = ['Accept-Encoding']
\n if content_type.startswith('text\/html'):
\n vary_headers.append('User-Agent')<\/p>\n
headers['Vary'] = ', '.join(vary_headers)<\/p>\n
# CDN\u7279\u5b9a\u5934\u90e8
\n headers['X-CDN-Cacheable'] = 'YES' if ttl > 0 else 'NO'
\n headers['X-CDN-TTL'] = str(ttl)<\/p>\n
return headers<\/p>\n
def analyze_cache_performance(self,
\n access_logs: List[Dict]) -> Dict[str, any]:
\n """\u5206\u6790\u7f13\u5b58\u6027\u80fd"""
\n # \u521d\u59cb\u5316\u7edf\u8ba1
\n total_by_status = {}
\n hit_by_status = {}
\n miss_by_status = {}<\/p>\n
for log in access_logs:
\n status = log.get('status_code')
\n cache_status = log.get('cache_status', 'MISS')<\/p>\n
if status not in total_by_status:
\n total_by_status[status] = 0
\n hit_by_status[status] = 0
\n miss_by_status[status] = 0<\/p>\n
total_by_status[status] += 1<\/p>\n
if cache_status == 'HIT':
\n hit_by_status[status] += 1
\n else:
\n miss_by_status[status] += 1<\/p>\n
# \u8ba1\u7b97\u547d\u4e2d\u7387
\n hit_rate_by_status = {}
\n for status in total_by_status:
\n total = total_by_status[status]
\n hits = hit_by_status.get(status, 0)
\n hit_rate = hits \/ total if total > 0 else 0
\n hit_rate_by_status[status] = hit_rate<\/p>\n
# \u8bc6\u522b\u4f18\u5316\u673a\u4f1a
\n optimization_opportunities = self._identify_optimization_opportunities(
\n hit_rate_by_status, total_by_status
\n )<\/p>\n
# \u8ba1\u7b97\u6f5c\u5728\u6536\u76ca
\n potential_savings = self._calculate_potential_savings(
\n total_by_status, hit_rate_by_status, optimization_opportunities
\n )<\/p>\n
return {
\n 'total_requests_by_status': total_by_status,
\n 'hit_rate_by_status': hit_rate_by_status,
\n 'optimization_opportunities': optimization_opportunities,
\n 'potential_savings': potential_savings,
\n 'current_configuration': {
\n 'cache_ttl': self.edge_config.cache_ttl,
\n 'stale_while_revalidate': self.edge_config.stale_while_revalidate,
\n 'stale_if_error': self.edge_config.stale_if_error
\n }
\n }<\/p>\n
def _identify_optimization_opportunities(self,
\n hit_rate_by_status: Dict[int, float],
\n total_by_status: Dict[int, int]) -> List[Dict]:
\n """\u8bc6\u522b\u4f18\u5316\u673a\u4f1a"""
\n opportunities = []<\/p>\n
for status, hit_rate in hit_rate_by_status.items():
\n total = total_by_status[status]
\n current_ttl = self.edge_config.cache_ttl.get(status, 0)<\/p>\n
# \u4f4e\u547d\u4e2d\u7387\u4f46\u9ad8\u9891\u7387
\n if hit_rate < 0.3 and total > 100:
\n opportunities.append({
\n 'status_code': status,
\n 'current_hit_rate': hit_rate,
\n 'request_count': total,
\n 'current_ttl': current_ttl,
\n 'issue': 'Low cache hit rate despite high traffic',
\n 'suggestion': 'Consider increasing TTL or reviewing cache keys'
\n })<\/p>\n
# \u9ad8\u547d\u4e2d\u7387\u4f46\u77edTTL
\n elif hit_rate > 0.8 and current_ttl < 300 and total > 50:
\n opportunities.append({
\n 'status_code': status,
\n 'current_hit_rate': hit_rate,
\n 'request_count': total,
\n 'current_ttl': current_ttl,
\n 'issue': 'High hit rate with short TTL',
\n 'suggestion': f"Increase TTL from {current_ttl}s to {current_ttl * 4}s"
\n })<\/p>\n
# \u53ef\u7f13\u5b58\u72b6\u6001\u7801\u4f46\u5f53\u524d\u4e0d\u7f13\u5b58
\n elif current_ttl == 0 and self._should_be_cacheable(status, total):
\n opportunities.append({
\n 'status_code': status,
\n 'current_hit_rate': hit_rate,
\n 'request_count': total,
\n 'current_ttl': current_ttl,
\n 'issue': 'Cacheable status code not being cached',
\n 'suggestion': f'Enable caching with TTL={self._suggest_ttl(status, total)}s'
\n })<\/p>\n
return sorted(opportunities,
\n key=lambda x: x['request_count'],
\n reverse=True)<\/p>\n
def _should_be_cacheable(self, status_code: int, request_count: int) -> bool:
\n """\u68c0\u67e5\u662f\u5426\u5e94\u8be5\u7f13\u5b58"""
\n # 404\u9519\u8bef\u5982\u679c\u9891\u7e41\u51fa\u73b0\u4e14\u5185\u5bb9\u76f8\u540c，\u53ef\u4ee5\u7f13\u5b58
\n if status_code == 404 and request_count > 100:
\n return True<\/p>\n
# \u81ea\u5b9a\u4e49\u72b6\u6001\u7801
\n if 450 <= status_code <= 499 or 520 <= status_code <= 599:
\n return request_count > 50<\/p>\n
return False<\/p>\n
def _suggest_ttl(self, status_code: int, request_count: int) -> int:
\n """\u5efa\u8baeTTL"""
\n if status_code == 404:
\n return 60 # 1\u5206\u949f
\n elif 450 <= status_code <= 499:
\n return 300 # 5\u5206\u949f
\n elif 520 <= status_code <= 599:
\n return 30 # 30\u79d2
\n else:
\n return 300 # \u9ed8\u8ba45\u5206\u949f<\/p>\n
def _calculate_potential_savings(self,
\n total_by_status: Dict[int, int],
\n hit_rate_by_status: Dict[int, float],
\n opportunities: List[Dict]) -> Dict[str, float]:
\n """\u8ba1\u7b97\u6f5c\u5728\u8282\u7701"""
\n # \u4f30\u7b97\u5e73\u5747\u54cd\u5e94\u5927\u5c0f（\u5b57\u8282）
\n avg_size_by_status = {
\n 200: 50000, # 50KB
\n 201: 1000, # 1KB
\n 204: 100, # 100B
\n 301: 500, # 500B
\n 302: 500,
\n 304: 100,
\n 404: 2000, # 2KB
\n 500: 5000, # 5KB
\n 460: 1000, # 1KB
\n 520: 500 # 500B
\n }<\/p>\n
current_bandwidth = 0
\n optimized_bandwidth = 0<\/p>\n
for status, total in total_by_status.items():
\n hit_rate = hit_rate_by_status.get(status, 0)
\n avg_size = avg_size_by_status.get(status, 1000)<\/p>\n
# \u5f53\u524d\u5e26\u5bbd\u4f7f\u7528
\n misses = total * (1 – hit_rate)
\n current_bandwidth += misses * avg_size<\/p>\n
# \u4f18\u5316\u540e\u5e26\u5bbd\u4f7f\u7528（\u5047\u8bbe\u547d\u4e2d\u7387\u63d0\u9ad8）
\n optimized_hit_rate = hit_rate<\/p>\n
# \u5e94\u7528\u4f18\u5316\u673a\u4f1a
\n for opp in opportunities:
\n if opp['status_code'] == status:
\n if 'increase' in opp['suggestion'].lower():
\n # \u5047\u8bbeTTL\u589e\u52a0\u4f1a\u4f7f\u547d\u4e2d\u7387\u63d0\u9ad8
\n optimized_hit_rate = min(hit_rate + 0.3, 0.95)
\n elif 'enable' in opp['suggestion'].lower():
\n # \u65b0\u542f\u7528\u7f13\u5b58
\n optimized_hit_rate = 0.7
\n break<\/p>\n
optimized_misses = total * (1 – optimized_hit_rate)
\n optimized_bandwidth += optimized_misses * avg_size<\/p>\n
bandwidth_savings = current_bandwidth – optimized_bandwidth
\n percentage_savings = (bandwidth_savings \/ current_bandwidth * 100
\n if current_bandwidth > 0 else 0)<\/p>\n
# \u5ef6\u8fdf\u8282\u7701\u4f30\u7b97
\n avg_origin_latency = 200 # ms
\n latency_savings = (current_bandwidth – optimized_bandwidth) \/ 1000 * avg_origin_latency<\/p>\n
return {
\n 'bandwidth_savings_bytes': bandwidth_savings,
\n 'bandwidth_savings_percentage': percentage_savings,
\n 'estimated_latency_reduction_ms': latency_savings,
\n 'estimated_cost_savings': self._estimate_cost_savings(bandwidth_savings)
\n }<\/p>\n
def _estimate_cost_savings(self, bandwidth_savings: float) -> float:
\n """\u4f30\u7b97\u6210\u672c\u8282\u7701"""
\n # \u5047\u8bbe $0.085 per GB (AWS CloudFront\u4ef7\u683c\u793a\u4f8b)
\n cost_per_gb = 0.085
\n savings_gb = bandwidth_savings \/ (1024 ** 3)
\n return savings_gb * cost_per_gb<\/p>\n
def optimize_edge_configuration(self,
\n analysis: Dict) -> CDNEdgeConfig:
\n """\u4f18\u5316\u8fb9\u7f18\u914d\u7f6e"""
\n optimized_ttl = self.edge_config.cache_ttl.copy()<\/p>\n
# \u6839\u636e\u5206\u6790\u8c03\u6574TTL
\n for opp in analysis['optimization_opportunities']:
\n status = opp['status_code']
\n suggestion = opp['suggestion']<\/p>\n
if 'Increase TTL' in suggestion:
\n # \u89e3\u6790\u5efa\u8bae\u7684\u65b0TTL
\n import re
\n match = re.search(r'to (\\\\d+)s', suggestion)
\n if match:
\n new_ttl = int(match.group(1))
\n optimized_ttl[status] = new_ttl<\/p>\n
elif 'Enable caching' in suggestion:
\n # \u542f\u7528\u7f13\u5b58
\n match = re.search(r'TTL=(\\\\d+)', suggestion)
\n if match:
\n new_ttl = int(match.group(1))
\n optimized_ttl[status] = new_ttl<\/p>\n
# \u521b\u5efa\u4f18\u5316\u540e\u7684\u914d\u7f6e
\n optimized_config = CDNEdgeConfig(
\n location=self.edge_config.location,
\n cache_ttl=optimized_ttl,
\n stale_while_revalidate=self.edge_config.stale_while_revalidate,
\n stale_if_error=self.edge_config.stale_if_error,
\n compression_enabled=self.edge_config.compression_enabled,
\n brotli_enabled=self.edge_config.brotli_enabled
\n )<\/p>\n
return optimized_config<\/p>\n
def generate_purge_strategy(self,
\n status_code_changes: Dict[int, str],
\n content_types: Set[str]) -> Dict[str, any]:
\n """\u751f\u6210\u6e05\u9664\u7b56\u7565"""
\n # \u4e0d\u540c\u72b6\u6001\u7801\u53d8\u5316\u7684\u6e05\u9664\u7b56\u7565
\n purge_strategies = []<\/p>\n
for status_code, change_type in status_code_changes.items():
\n if change_type == 'error_to_success':
\n # \u4f8b\u5982 404 -> 200
\n strategy = {
\n 'status_code': status_code,
\n 'change': change_type,
\n 'purge_scope': 'specific_urls',
\n 'purge_method': 'immediate',
\n 'priority': 'high',
\n 'estimated_impact': 'Users will see updated content immediately'
\n }<\/p>\n
elif change_type == 'success_to_error':
\n # \u4f8b\u5982 200 -> 404
\n strategy = {
\n 'status_code': status_code,
\n 'change': change_type,
\n 'purge_scope': 'specific_urls',
\n 'purge_method': 'immediate',
\n 'priority': 'critical',
\n 'estimated_impact': 'Prevent users from accessing deleted content'
\n }<\/p>\n
elif change_type == 'custom_code_change':
\n # \u4f8b\u5982 460 -> 200（\u5546\u54c1\u91cd\u65b0\u4e0a\u67b6）
\n strategy = {
\n 'status_code': status_code,
\n 'change': change_type,
\n 'purge_scope': 'specific_urls',
\n 'purge_method': 'immediate',
\n 'priority': 'medium',
\n 'estimated_impact': 'Update business status immediately'
\n }<\/p>\n
else:
\n strategy = {
\n 'status_code': status_code,
\n 'change': change_type,
\n 'purge_scope': 'pattern',
\n 'purge_method': 'scheduled',
\n 'priority': 'low'
\n }<\/p>\n
purge_strategies.append(strategy)<\/p>\n
# \u57fa\u4e8e\u5185\u5bb9\u7c7b\u578b\u7684\u6e05\u9664\u5efa\u8bae
\n content_type_purge = {}
\n for content_type in content_types:
\n if content_type.startswith('text\/html'):
\n content_type_purge[content_type] = {
\n 'cache_duration': 'short',
\n 'purge_frequency': 'high',
\n 'recommendation': 'Use surrogate keys for granular purging'
\n }
\n elif 'image' in content_type:
\n content_type_purge[content_type] = {
\n 'cache_duration': 'long',
\n 'purge_frequency': 'low',
\n 'recommendation': 'Use versioned URLs for cache busting'
\n }
\n elif 'application\/json' in content_type:
\n content_type_purge[content_type] = {
\n 'cache_duration': 'medium',
\n 'purge_frequency': 'medium',
\n 'recommendation': 'Implement cache revalidation with ETags'
\n }<\/p>\n
return {
\n 'purge_strategies': purge_strategies,
\n 'content_type_recommendations': content_type_purge,
\n 'best_practices': self._get_purge_best_practices()
\n }<\/p>\n
def _get_purge_best_practices(self) -> List[str]:
\n """\u83b7\u53d6\u6e05\u9664\u6700\u4f73\u5b9e\u8df5"""
\n return [
\n "Use surrogate keys for granular cache purging",
\n "Implement soft purge (stale-while-revalidate) when possible",
\n "Schedule bulk purges during low-traffic periods",
\n "Monitor purge queue depth to avoid delays",
\n "Use cache tags for related content invalidation",
\n "Implement purge ACLs to prevent accidental purges"
\n ]<\/p>\n
# CDN\u6027\u80fd\u76d1\u63a7
\nclass CDNPerformanceMonitor:
\n """CDN\u6027\u80fd\u76d1\u63a7\u5668"""<\/p>\n
def __init__(self, optimizer: CDNOptimizer):
\n self.optimizer = optimizer
\n self.metrics_history = []
\n self.alert_thresholds = {
\n 'hit_rate': 0.3, # \u4f4e\u4e8e30%\u544a\u8b66
\n 'origin_load': 0.8, # \u6e90\u7ad9\u8d1f\u8f7d\u8d85\u8fc780%\u544a\u8b66
\n 'error_rate': 0.05, # \u9519\u8bef\u7387\u8d85\u8fc75%\u544a\u8b66
\n 'latency_p95': 1000 # P95\u5ef6\u8fdf\u8d85\u8fc71\u79d2\u544a\u8b66
\n }<\/p>\n
def record_metrics(self, metrics: Dict[str, any]):
\n """\u8bb0\u5f55\u6307\u6807"""
\n self.metrics_history.append({
\n 'timestamp': datetime.now(),
\n 'metrics': metrics
\n })<\/p>\n
# \u4fdd\u6301\u5386\u53f2\u6570\u636e\u5927\u5c0f
\n if len(self.metrics_history) > 1000:
\n self.metrics_history = self.metrics_history[-1000:]<\/p>\n
# \u68c0\u67e5\u544a\u8b66
\n alerts = self._check_alerts(metrics)
\n if alerts:
\n self._trigger_alerts(alerts)<\/p>\n
def _check_alerts(self, metrics: Dict) -> List[Dict]:
\n """\u68c0\u67e5\u544a\u8b66"""
\n alerts = []<\/p>\n
# \u68c0\u67e5\u547d\u4e2d\u7387
\n hit_rate = metrics.get('overall_hit_rate', 0)
\n if hit_rate < self.alert_thresholds['hit_rate']:
\n alerts.append({
\n 'type': 'LOW_HIT_RATE',
\n 'value': hit_rate,
\n 'threshold': self.alert_thresholds['hit_rate'],
\n 'severity': 'WARNING'
\n })<\/p>\n
# \u68c0\u67e5\u6e90\u7ad9\u8d1f\u8f7d
\n origin_load = metrics.get('origin_load_percentage', 0)
\n if origin_load > self.alert_thresholds['origin_load']:
\n alerts.append({
\n 'type': 'HIGH_ORIGIN_LOAD',
\n 'value': origin_load,
\n 'threshold': self.alert_thresholds['origin_load'],
\n 'severity': 'CRITICAL'
\n })<\/p>\n
# \u68c0\u67e5\u9519\u8bef\u7387
\n error_rate = metrics.get('error_rate', 0)
\n if error_rate > self.alert_thresholds['error_rate']:
\n alerts.append({
\n 'type': 'HIGH_ERROR_RATE',
\n 'value': error_rate,
\n 'threshold': self.alert_thresholds['error_rate'],
\n 'severity': 'ERROR'
\n })<\/p>\n
# \u68c0\u67e5\u5ef6\u8fdf
\n latency_p95 = metrics.get('latency_p95_ms', 0)
\n if latency_p95 > self.alert_thresholds['latency_p95']:
\n alerts.append({
\n 'type': 'HIGH_LATENCY',
\n 'value': latency_p95,
\n 'threshold': self.alert_thresholds['latency_p95'],
\n 'severity': 'WARNING'
\n })<\/p>\n
return alerts<\/p>\n
def _trigger_alerts(self, alerts: List[Dict]):
\n """\u89e6\u53d1\u544a\u8b66"""
\n for alert in alerts:
\n print(f"CDN ALERT [{alert['severity']}]: {alert['type']} = {alert['value']}")<\/p>\n
def generate_performance_report(self,
\n start_time: datetime,
\n end_time: datetime) -> Dict[str, any]:
\n """\u751f\u6210\u6027\u80fd\u62a5\u544a"""
\n # \u8fc7\u6ee4\u65f6\u95f4\u8303\u56f4\u5185\u7684\u6307\u6807
\n relevant_metrics = [
\n m for m in self.metrics_history
\n if start_time <= m['timestamp'] <= end_time
\n ]<\/p>\n
if not relevant_metrics:
\n return {}<\/p>\n
# \u8ba1\u7b97\u7edf\u8ba1
\n hit_rates = [m['metrics'].get('overall_hit_rate', 0) for m in relevant_metrics]
\n error_rates = [m['metrics'].get('error_rate', 0) for m in relevant_metrics]
\n latencies = [m['metrics'].get('latency_p95_ms', 0) for m in relevant_metrics]<\/p>\n
import statistics<\/p>\n
report = {
\n 'time_period': {
\n 'start': start_time,
\n 'end': end_time
\n },
\n 'summary': {
\n 'avg_hit_rate': statistics.mean(hit_rates),
\n 'avg_error_rate': statistics.mean(error_rates),
\n 'avg_latency_p95': statistics.mean(latencies),
\n 'total_alerts': sum(len(m.get('alerts', [])) for m in relevant_metrics)
\n },
\n 'trends': self._analyze_trends(relevant_metrics),
\n 'recommendations': self._generate_recommendations(relevant_metrics)
\n }<\/p>\n
return report<\/p>\n
def _analyze_trends(self, metrics_data: List[Dict]) -> Dict[str, any]:
\n """\u5206\u6790\u8d8b\u52bf"""
\n # \u7b80\u5316\u5b9e\u73b0
\n return {
\n 'hit_rate_trend': 'stable',
\n 'latency_trend': 'stable',
\n 'error_rate_trend': 'stable'
\n }<\/p>\n
def _generate_recommendations(self, metrics_data: List[Dict]) -> List[Dict]:
\n """\u751f\u6210\u63a8\u8350"""
\n recommendations = []<\/p>\n
# \u5206\u6790\u5e73\u5747\u547d\u4e2d\u7387
\n avg_hit_rate = statistics.mean(
\n [m['metrics'].get('overall_hit_rate', 0) for m in metrics_data]
\n )<\/p>\n
if avg_hit_rate < 0.5:
\n recommendations.append({
\n 'type': 'CACHE_OPTIMIZATION',
\n 'priority': 'HIGH',
\n 'description': f'Low cache hit rate ({avg_hit_rate:.1%}). Consider optimizing cache TTLs and keys.',
\n 'action': 'Review and adjust CDN cache configuration'
\n })<\/p>\n
# \u5206\u6790\u9519\u8bef\u6a21\u5f0f
\n error_details = []
\n for data in metrics_data:
\n errors = data['metrics'].get('errors_by_status', {})
\n error_details.extend(errors.items())<\/p>\n
# \u627e\u51fa\u6700\u5e38\u89c1\u7684\u9519\u8bef\u72b6\u6001\u7801
\n from collections import Counter
\n error_counter = Counter()
\n for status, count in error_details:
\n error_counter[status] += count<\/p>\n
common_errors = error_counter.most_common(3)
\n for status, count in common_errors:
\n if status >= 500:
\n recommendations.append({
\n 'type': 'ERROR_RESOLUTION',
\n 'priority': 'CRITICAL',
\n 'description': f'High server errors ({status}: {count} occurrences). Investigate backend issues.',
\n 'action': f'Debug status code {status} errors'
\n })<\/p>\n
return recommendations<\/p>\n
# \u4f7f\u7528\u793a\u4f8b
\nedge_config = CDNEdgeConfig.default_config()
\ncdn_optimizer = CDNOptimizer(edge_config)
\nmonitor = CDNPerformanceMonitor(cdn_optimizer)<\/p>\n
# \u6a21\u62dfCDN\u8bbf\u95ee\u65e5\u5fd7
\naccess_logs = []
\nfor i in range(1000):
\n status_code = 200 if i % 10 != 0 else 404
\n cache_status = 'HIT' if i % 3 != 0 else 'MISS'<\/p>\n
access_logs.append({
\n 'status_code': status_code,
\n 'cache_status': cache_status,
\n 'url': f'\/api\/products\/{i % 100}',
\n 'response_size': 5000 if status_code == 200 else 1000
\n })<\/p>\n
# \u5206\u6790\u7f13\u5b58\u6027\u80fd
\nanalysis = cdn_optimizer.analyze_cache_performance(access_logs)<\/p>\n
print("=== CDN Cache Analysis ===")
\nprint(f"Overall hit rate: {analysis['hit_rate_by_status'].get(200, 0):.1%}")
\nprint(f"404 hit rate: {analysis['hit_rate_by_status'].get(404, 0):.1%}")<\/p>\n
print("\\\\nOptimization opportunities:")
\nfor opp in analysis['optimization_opportunities'][:3]:
\n print(f" Status {opp['status_code']}: {opp['issue']}")<\/p>\n
print(f"\\\\nPotential bandwidth savings: {analysis['potential_savings']['bandwidth_savings_percentage']:.1f}%")
\nprint(f"Estimated cost savings: ${analysis['potential_savings']['estimated_cost_savings']:.2f}")<\/p>\n
# \u751f\u6210\u4f18\u5316\u914d\u7f6e
\noptimized_config = cdn_optimizer.optimize_edge_configuration(analysis)
\nprint(f"\\\\nOptimized 404 TTL: {optimized_config.cache_ttl.get(404)}s (was: {edge_config.cache_ttl.get(404)}s)")<\/p>\n
# \u76d1\u63a7\u6027\u80fd
\nmonitor.record_metrics({
\n 'overall_hit_rate': 0.65,
\n 'origin_load_percentage': 0.75,
\n 'error_rate': 0.02,
\n 'latency_p95_ms': 850,
\n 'errors_by_status': {404: 10, 500: 2}
\n})<\/p>\n
# \u751f\u6210\u62a5\u544a
\nreport = monitor.generate_performance_report(
\n start_time=datetime.now() – timedelta(hours=1),
\n end_time=datetime.now()
\n)<\/p>\n
print(f"\\\\nPerformance report:")
\nprint(f"Average hit rate: {report['summary']['avg_hit_rate']:.1%}")
\nprint(f"Recommendations: {len(report['recommendations'])}")<\/p>\n
python<\/p>\n
# \u81ea\u9002\u5e94\u72b6\u6001\u7801\u54cd\u5e94\u7cfb\u7edf
\nfrom typing import Dict, Any, Optional, Callable
\nfrom dataclasses import dataclass
\nfrom datetime import datetime, timedelta
\nimport time
\nimport random<\/p>\n
class AdaptiveResponseStrategy:
\n """\u81ea\u9002\u5e94\u54cd\u5e94\u7b56\u7565"""<\/p>\n
def __init__(self):
\n self.strategies = self._initialize_strategies()
\n self.performance_history = []
\n self.learning_rate = 0.1<\/p>\n
def _initialize_strategies(self) -> Dict[str, Dict[str, Any]]:
\n """\u521d\u59cb\u5316\u7b56\u7565"""
\n return {
\n 'aggressive_caching': {
\n 'description': 'Aggressive caching for high traffic',
\n 'conditions': ['high_traffic', 'low_error_rate'],
\n 'actions': {
\n 'increase_cache_ttl': 2.0, # \u53cc\u500dTTL
\n 'enable_stale_while_revalidate': True,
\n 'prefetch_related': True
\n },
\n 'performance_impact': 0.0 # \u5f85\u5b66\u4e60
\n },
\n 'conservative_fallback': {
\n 'description': 'Conservative fallback during instability',
\n 'conditions': ['high_error_rate', 'increasing_latency'],
\n 'actions': {
\n 'reduce_cache_ttl': 0.5, # \u4e00\u534aTTL
\n 'disable_complex_features': True,
\n 'enable_circuit_breaker': True
\n },
\n 'performance_impact': 0.0
\n },
\n 'graceful_degradation': {
\n 'description': 'Graceful degradation for overload',
\n 'conditions': ['high_load', 'resource_constrained'],
\n 'actions': {
\n 'simplify_responses': True,
\n 'return_partial_data': True,
\n 'suggest_retry_later': True
\n },
\n 'performance_impact': 0.0
\n },
\n 'predictive_optimization': {
\n 'description': 'Predictive optimization based on patterns',
\n 'conditions': ['stable_pattern', 'predictable_load'],
\n 'actions': {
\n 'pre_warm_cache': True,
\n 'schedule_maintenance': True,
\n 'optimize_anticipatory': True
\n },
\n 'performance_impact': 0.0
\n }
\n }<\/p>\n
def analyze_context(self,
\n metrics: Dict[str, Any],
\n request_context: Dict[str, Any]) -> List[str]:
\n """\u5206\u6790\u4e0a\u4e0b\u6587，\u786e\u5b9a\u9002\u7528\u7b56\u7565"""
\n applicable_strategies = []<\/p>\n
# \u68c0\u67e5\u5404\u9879\u6761\u4ef6
\n conditions = self._evaluate_conditions(metrics, request_context)<\/p>\n
for strategy_name, strategy in self.strategies.items():
\n required_conditions = strategy['conditions']<\/p>\n
# \u68c0\u67e5\u662f\u5426\u6ee1\u8db3\u6240\u6709\u6761\u4ef6
\n if all(cond in conditions for cond in required_conditions):
\n applicable_strategies.append(strategy_name)<\/p>\n
return applicable_strategies<\/p>\n
def _evaluate_conditions(self,
\n metrics: Dict[str, Any],
\n context: Dict[str, Any]) -> List[str]:
\n """\u8bc4\u4f30\u6761\u4ef6"""
\n conditions = []<\/p>\n
# \u6d41\u91cf\u6761\u4ef6
\n request_rate = metrics.get('requests_per_second', 0)
\n if request_rate > 100:
\n conditions.append('high_traffic')
\n elif request_rate < 10:
\n conditions.append('low_traffic')<\/p>\n
# \u9519\u8bef\u7387\u6761\u4ef6
\n error_rate = metrics.get('error_rate', 0)
\n if error_rate > 0.1:
\n conditions.append('high_error_rate')
\n elif error_rate < 0.01:
\n conditions.append('low_error_rate')<\/p>\n
# \u5ef6\u8fdf\u6761\u4ef6
\n latency_trend = metrics.get('latency_trend', 'stable')
\n if latency_trend == 'increasing':
\n conditions.append('increasing_latency')<\/p>\n
# \u8d1f\u8f7d\u6761\u4ef6
\n system_load = metrics.get('system_load', 0)
\n if system_load > 0.8:
\n conditions.append('high_load')<\/p>\n
# \u8d44\u6e90\u6761\u4ef6
\n memory_usage = metrics.get('memory_usage', 0)
\n if memory_usage > 0.9:
\n conditions.append('resource_constrained')<\/p>\n
# \u6a21\u5f0f\u6761\u4ef6
\n if context.get('pattern_stability', 0) > 0.8:
\n conditions.append('stable_pattern')<\/p>\n
if context.get('load_predictability', 0) > 0.7:
\n conditions.append('predictable_load')<\/p>\n
return conditions<\/p>\n
def select_best_strategy(self,
\n applicable_strategies: List[str],
\n historical_performance: Dict[str, float]) -> str:
\n """\u9009\u62e9\u6700\u4f73\u7b56\u7565"""
\n if not applicable_strategies:
\n return 'default'<\/p>\n
# \u4f7f\u7528UCB（Upper Confidence Bound）\u7b97\u6cd5\u5e73\u8861\u63a2\u7d22\u548c\u5229\u7528
\n strategies_with_score = []<\/p>\n
for strategy in applicable_strategies:
\n if strategy in historical_performance:
\n # \u5229\u7528：\u4f7f\u7528\u5386\u53f2\u6027\u80fd
\n avg_performance = historical_performance[strategy]['average']
\n count = historical_performance[strategy]['count']<\/p>\n
# \u63a2\u7d22\u9879：\u9f13\u52b1\u5c1d\u8bd5\u6b21\u6570\u5c11\u7684\u7b56\u7565
\n exploration_bonus = (2 * math.log(sum(
\n h['count'] for h in historical_performance.values()
\n )) \/ count) ** 0.5 if count > 0 else float('inf')<\/p>\n
score = avg_performance + exploration_bonus
\n else:
\n # \u65b0\u7b56\u7565，\u7ed9\u9ad8\u5206\u9f13\u52b1\u5c1d\u8bd5
\n score = float('inf')<\/p>\n
strategies_with_score.append((strategy, score))<\/p>\n
# \u9009\u62e9\u6700\u9ad8\u5206\u7b56\u7565
\n return max(strategies_with_score, key=lambda x: x[0])[0]<\/p>\n
def adapt_response(self,
\n original_status_code: int,
\n selected_strategy: str,
\n context: Dict[str, Any]) -> Dict[str, Any]:
\n """\u6839\u636e\u7b56\u7565\u8c03\u6574\u54cd\u5e94"""
\n if selected_strategy == 'default':
\n return {
\n 'status_code': original_status_code,
\n 'headers': {},
\n 'body': {},
\n 'strategy_applied': 'none'
\n }<\/p>\n
strategy = self.strategies[selected_strategy]
\n actions = strategy['actions']<\/p>\n
adapted_response = {
\n 'status_code': original_status_code,
\n 'headers': {},
\n 'body': {},
\n 'strategy_applied': selected_strategy,
\n 'actions_taken': []
\n }<\/p>\n
# \u5e94\u7528\u5404\u4e2a\u52a8\u4f5c
\n if 'increase_cache_ttl' in actions:
\n multiplier = actions['increase_cache_ttl']
\n adapted_response['headers']['Cache-Control'] = \\\\
\n f"public, max-age={3600 * multiplier}" # \u793a\u4f8b<\/p>\n
if 'reduce_cache_ttl' in actions:
\n multiplier = actions['reduce_cache_ttl']
\n adapted_response['headers']['Cache-Control'] = \\\\
\n f"public, max-age={300 * multiplier}"<\/p>\n
if 'enable_stale_while_revalidate' in actions:
\n adapted_response['headers']['Cache-Control'] += \\\\
\n ", stale-while-revalidate=300"<\/p>\n
if 'simplify_responses' in actions and original_status_code == 200:
\n # \u7b80\u5316\u54cd\u5e94\u4f53
\n adapted_response['body'] = {
\n 'simplified': True,
\n 'essential_data_only': True
\n }<\/p>\n
if 'return_partial_data' in actions and original_status_code == 200:
\n # \u8fd4\u56de\u90e8\u5206\u6570\u636e
\n adapted_response['status_code'] = 206 # Partial Content
\n adapted_response['headers']['Content-Range'] = 'bytes 0-499\/*'<\/p>\n
if 'suggest_retry_later' in actions:
\n # \u5efa\u8bae\u7a0d\u540e\u91cd\u8bd5
\n retry_after = context.get('estimated_recovery_seconds', 30)
\n adapted_response['headers']['Retry-After'] = str(retry_after)<\/p>\n
# \u5bf9\u4e8e\u6210\u529f\u72b6\u6001\u7801，\u6dfb\u52a0\u8b66\u544a
\n if 200 <= original_status_code < 300:
\n adapted_response['headers']['Warning'] = \\\\
\n '199 – "Service experiencing high load"'<\/p>\n
return adapted_response<\/p>\n
def record_performance(self,
\n strategy: str,
\n performance_metrics: Dict[str, float]):
\n """\u8bb0\u5f55\u7b56\u7565\u6027\u80fd"""
\n self.performance_history.append({
\n 'timestamp': datetime.now(),
\n 'strategy': strategy,
\n 'metrics': performance_metrics
\n })<\/p>\n
# \u66f4\u65b0\u7b56\u7565\u6027\u80fd
\n if strategy in self.strategies:
\n # \u8ba1\u7b97\u6027\u80fd\u5206\u6570
\n performance_score = self._calculate_performance_score(performance_metrics)<\/p>\n
# \u66f4\u65b0\u5b66\u4e60
\n if 'performance_impact' in self.strategies[strategy]:
\n current = self.strategies[strategy]['performance_impact']
\n updated = current + self.learning_rate * (performance_score – current)
\n self.strategies[strategy]['performance_impact'] = updated<\/p>\n
def _calculate_performance_score(self, metrics: Dict[str, float]) -> float:
\n """\u8ba1\u7b97\u6027\u80fd\u5206\u6570"""
\n # \u7ec4\u5408\u591a\u4e2a\u6307\u6807
\n score = 0.0<\/p>\n
# \u54cd\u5e94\u65f6\u95f4（\u8d8a\u4f4e\u8d8a\u597d）
\n response_time = metrics.get('response_time_ms', 1000)
\n response_time_score = max(0, 1 – (response_time \/ 5000)) # 5\u79d2\u4e3a\u9608\u503c
\n score += response_time_score * 0.4<\/p>\n
# \u9519\u8bef\u7387（\u8d8a\u4f4e\u8d8a\u597d）
\n error_rate = metrics.get('error_rate', 0)
\n error_rate_score = max(0, 1 – (error_rate \/ 0.5)) # 50%\u4e3a\u9608\u503c
\n score += error_rate_score * 0.3<\/p>\n
# \u7f13\u5b58\u547d\u4e2d\u7387（\u8d8a\u9ad8\u8d8a\u597d）
\n cache_hit_rate = metrics.get('cache_hit_rate', 0)
\n score += cache_hit_rate * 0.2<\/p>\n
# \u8d44\u6e90\u4f7f\u7528（\u8d8a\u4f4e\u8d8a\u597d）
\n resource_usage = metrics.get('resource_usage', 0)
\n resource_score = max(0, 1 – resource_usage)
\n score += resource_score * 0.1<\/p>\n
return score<\/p>\n
def get_learning_report(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u5b66\u4e60\u62a5\u544a"""
\n # \u8ba1\u7b97\u7b56\u7565\u6027\u80fd\u7edf\u8ba1
\n strategy_stats = {}<\/p>\n
for strategy_name in self.strategies:
\n strategy_performances = [
\n entry for entry in self.performance_history
\n if entry['strategy'] == strategy_name
\n ]<\/p>\n
if strategy_performances:
\n scores = [
\n self._calculate_performance_score(p['metrics'])
\n for p in strategy_performances
\n ]<\/p>\n
import statistics
\n strategy_stats[strategy_name] = {
\n 'count': len(strategy_performances),
\n 'avg_score': statistics.mean(scores) if scores else 0,
\n 'std_score': statistics.stdev(scores) if len(scores) > 1 else 0,
\n 'last_used': max(
\n p['timestamp'] for p in strategy_performances
\n ) if strategy_performances else None
\n }<\/p>\n
# \u5b66\u4e60\u8d8b\u52bf
\n learning_trend = self._analyze_learning_trend()<\/p>\n
# \u63a8\u8350\u7b56\u7565
\n recommended_strategies = self._recommend_strategies(strategy_stats)<\/p>\n
return {
\n 'strategy_statistics': strategy_stats,
\n 'learning_trend': learning_trend,
\n 'recommended_strategies': recommended_strategies,
\n 'learning_efficiency': self._calculate_learning_efficiency()
\n }<\/p>\n
def _analyze_learning_trend(self) -> Dict[str, Any]:
\n """\u5206\u6790\u5b66\u4e60\u8d8b\u52bf"""
\n if len(self.performance_history) < 10:
\n return {'status': 'INSUFFICIENT_DATA'}<\/p>\n
# \u8ba1\u7b97\u6027\u80fd\u6539\u8fdb
\n recent_performance = self.performance_history[-10:]
\n older_performance = self.performance_history[:10]<\/p>\n
recent_scores = [
\n self._calculate_performance_score(p['metrics'])
\n for p in recent_performance
\n ]
\n older_scores = [
\n self._calculate_performance_score(p['metrics'])
\n for p in older_performance
\n ]<\/p>\n
import statistics
\n improvement = statistics.mean(recent_scores) – statistics.mean(older_scores)<\/p>\n
if improvement > 0.1:
\n trend = 'IMPROVING'
\n elif improvement < -0.1:
\n trend = 'DECLINING'
\n else:
\n trend = 'STABLE'<\/p>\n
return {
\n 'status': trend,
\n 'improvement': improvement,
\n 'confidence': min(1.0, len(self.performance_history) \/ 100)
\n }<\/p>\n
def _recommend_strategies(self,
\n strategy_stats: Dict[str, Dict]) -> List[Dict[str, Any]]:
\n """\u63a8\u8350\u7b56\u7565"""
\n recommendations = []<\/p>\n
for strategy_name, stats in strategy_stats.items():
\n avg_score = stats['avg_score']<\/p>\n
if avg_score > 0.8:
\n recommendations.append({
\n 'strategy': strategy_name,
\n 'recommendation': 'HIGHLY_EFFECTIVE',
\n 'confidence': 'HIGH',
\n 'suggested_usage': 'Increase application frequency'
\n })
\n elif avg_score < 0.3:
\n recommendations.append({
\n 'strategy': strategy_name,
\n 'recommendation': 'INEFFECTIVE',
\n 'confidence': 'MEDIUM',
\n 'suggested_usage': 'Review and potentially deprecate'
\n })<\/p>\n
return sorted(recommendations,
\n key=lambda x: x.get('confidence', 'LOW'),
\n reverse=True)<\/p>\n
def _calculate_learning_efficiency(self) -> float:
\n """\u8ba1\u7b97\u5b66\u4e60\u6548\u7387"""
\n if len(self.performance_history) < 20:
\n return 0.0<\/p>\n
# \u8ba1\u7b97\u6027\u80fd\u65b9\u5dee\u51cf\u5c11
\n early_scores = [
\n self._calculate_performance_score(p['metrics'])
\n for p in self.performance_history[:10]
\n ]
\n late_scores = [
\n self._calculate_performance_score(p['metrics'])
\n for p in self.performance_history[-10:]
\n ]<\/p>\n
import statistics
\n early_variance = statistics.variance(early_scores) if len(early_scores) > 1 else 1
\n late_variance = statistics.variance(late_scores) if len(late_scores) > 1 else 1<\/p>\n
# \u65b9\u5dee\u51cf\u5c11\u8868\u793a\u5b66\u4e60\u6548\u7387
\n variance_reduction = (early_variance – late_variance) \/ early_variance
\n return max(0, min(1, variance_reduction))<\/p>\n
# \u81ea\u9002\u5e94\u4e2d\u95f4\u4ef6
\nclass AdaptiveStatusCodeMiddleware:
\n """\u81ea\u9002\u5e94\u72b6\u6001\u7801\u4e2d\u95f4\u4ef6"""<\/p>\n
def __init__(self, strategy_engine: AdaptiveResponseStrategy):
\n self.strategy_engine = strategy_engine
\n self.metrics_collector = MetricsCollector()
\n self.context_analyzer = ContextAnalyzer()<\/p>\n
async def process_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
\n """\u5904\u7406\u8bf7\u6c42"""
\n # \u6536\u96c6\u6307\u6807
\n current_metrics = self.metrics_collector.get_current_metrics()<\/p>\n
# \u5206\u6790\u4e0a\u4e0b\u6587
\n request_context = self.context_analyzer.analyze(request)<\/p>\n
# \u9009\u62e9\u7b56\u7565
\n applicable_strategies = self.strategy_engine.analyze_context(
\n current_metrics, request_context
\n )<\/p>\n
historical_performance = self._get_historical_performance()
\n selected_strategy = self.strategy_engine.select_best_strategy(
\n applicable_strategies, historical_performance
\n )<\/p>\n
# \u5b58\u50a8\u7b56\u7565\u51b3\u7b56
\n request['adaptive_strategy'] = selected_strategy
\n request['request_context'] = request_context<\/p>\n
return request<\/p>\n
async def process_response(self,
\n request: Dict[str, Any],
\n response: Dict[str, Any]) -> Dict[str, Any]:
\n """\u5904\u7406\u54cd\u5e94"""
\n selected_strategy = request.get('adaptive_strategy', 'default')
\n request_context = request.get('request_context', {})<\/p>\n
# \u81ea\u9002\u5e94\u8c03\u6574\u54cd\u5e94
\n adapted_response = self.strategy_engine.adapt_response(
\n response['status_code'],
\n selected_strategy,
\n request_context
\n )<\/p>\n
# \u5408\u5e76\u54cd\u5e94
\n final_response = response.copy()
\n final_response.update({
\n 'headers': {**response.get('headers', {}),
\n **adapted_response.get('headers', {})},
\n 'body': adapted_response.get('body') or response.get('body'),
\n 'adaptive_metadata': {
\n 'strategy_applied': adapted_response.get('strategy_applied'),
\n 'actions_taken': adapted_response.get('actions_taken', [])
\n }
\n })<\/p>\n
# \u8bb0\u5f55\u6027\u80fd
\n self._record_performance(selected_strategy, request, final_response)<\/p>\n
return final_response<\/p>\n
def _get_historical_performance(self) -> Dict[str, Dict[str, float]]:
\n """\u83b7\u53d6\u5386\u53f2\u6027\u80fd"""
\n # \u7b80\u5316\u5b9e\u73b0
\n return {}<\/p>\n
def _record_performance(self,
\n strategy: str,
\n request: Dict[str, Any],
\n response: Dict[str, Any]):
\n """\u8bb0\u5f55\u6027\u80fd"""
\n performance_metrics = {
\n 'response_time_ms': response.get('response_time', 0),
\n 'error_rate': 1 if response['status_code'] >= 400 else 0,
\n 'cache_hit_rate': 1 if response.get('from_cache', False) else 0,
\n 'resource_usage': request.get('resource_usage', 0)
\n }<\/p>\n
self.strategy_engine.record_performance(strategy, performance_metrics)<\/p>\n
def get_adaptation_report(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u9002\u5e94\u62a5\u544a"""
\n learning_report = self.strategy_engine.get_learning_report()
\n current_metrics = self.metrics_collector.get_current_metrics()<\/p>\n
return {
\n 'learning_report': learning_report,
\n 'current_metrics': current_metrics,
\n 'adaptation_efficiency': self._calculate_adaptation_efficiency(),
\n 'recommended_adjustments': self._generate_adjustment_recommendations(
\n learning_report, current_metrics
\n )
\n }<\/p>\n
def _calculate_adaptation_efficiency(self) -> float:
\n """\u8ba1\u7b97\u9002\u5e94\u6548\u7387"""
\n # \u57fa\u4e8e\u5b66\u4e60\u6548\u7387\u548c\u5f53\u524d\u6027\u80fd
\n learning_report = self.strategy_engine.get_learning_report()
\n learning_efficiency = learning_report.get('learning_efficiency', 0)<\/p>\n
current_metrics = self.metrics_collector.get_current_metrics()
\n current_performance = self.strategy_engine._calculate_performance_score(
\n current_metrics
\n )<\/p>\n
# \u7ec4\u5408\u6307\u6807
\n efficiency = (learning_efficiency * 0.3 + current_performance * 0.7)
\n return efficiency<\/p>\n
def _generate_adjustment_recommendations(self,
\n learning_report: Dict[str, Any],
\n current_metrics: Dict[str, Any]) -> List[Dict[str, Any]]:
\n """\u751f\u6210\u8c03\u6574\u5efa\u8bae"""
\n recommendations = []<\/p>\n
# \u57fa\u4e8e\u5b66\u4e60\u62a5\u544a\u7684\u5efa\u8bae
\n for rec in learning_report.get('recommended_strategies', []):
\n if rec['recommendation'] == 'INEFFECTIVE':
\n recommendations.append({
\n 'type': 'STRATEGY_ADJUSTMENT',
\n 'priority': 'MEDIUM',
\n 'description': f"Strategy '{rec['strategy']}' appears ineffective",
\n 'action': f"Review and adjust {rec['strategy']} strategy parameters"
\n })<\/p>\n
# \u57fa\u4e8e\u5f53\u524d\u6307\u6807\u7684\u5efa\u8bae
\n if current_metrics.get('error_rate', 0) > 0.1:
\n recommendations.append({
\n 'type': 'ERROR_HANDLING',
\n 'priority': 'HIGH',
\n 'description': 'High error rate detected',
\n 'action': 'Enable conservative fallback strategies more aggressively'
\n })<\/p>\n
if current_metrics.get('requests_per_second', 0) > 500:
\n recommendations.append({
\n 'type': 'SCALING_ADJUSTMENT',
\n 'priority': 'MEDIUM',
\n 'description': 'High traffic load',
\n 'action': 'Increase caching aggressiveness and enable prefetching'
\n })<\/p>\n
return recommendations<\/p>\n
# \u652f\u6301\u7c7b
\nclass MetricsCollector:
\n """\u6307\u6807\u6536\u96c6\u5668"""<\/p>\n
def get_current_metrics(self) -> Dict[str, float]:
\n """\u83b7\u53d6\u5f53\u524d\u6307\u6807"""
\n # \u7b80\u5316\u5b9e\u73b0
\n return {
\n 'requests_per_second': random.uniform(50, 500),
\n 'error_rate': random.uniform(0, 0.2),
\n 'latency_trend': random.choice(['stable', 'increasing', 'decreasing']),
\n 'system_load': random.uniform(0.3, 0.9),
\n 'memory_usage': random.uniform(0.4, 0.95)
\n }<\/p>\n
class ContextAnalyzer:
\n """\u4e0a\u4e0b\u6587\u5206\u6790\u5668"""<\/p>\n
def analyze(self, request: Dict[str, Any]) -> Dict[str, Any]:
\n """\u5206\u6790\u8bf7\u6c42\u4e0a\u4e0b\u6587"""
\n return {
\n 'pattern_stability': random.uniform(0.5, 1.0),
\n 'load_predictability': random.uniform(0.3, 0.9),
\n 'estimated_recovery_seconds': random.randint(10, 300),
\n 'resource_usage': random.uniform(0.2, 0.8)
\n }<\/p>\n
# \u4f7f\u7528\u793a\u4f8b
\nstrategy_engine = AdaptiveResponseStrategy()
\nmiddleware = AdaptiveStatusCodeMiddleware(strategy_engine)<\/p>\n
# \u6a21\u62df\u8bf7\u6c42\u5904\u7406
\nfor i in range(100):
\n request = {
\n 'method': 'GET',
\n 'url': f'\/api\/data\/{i}',
\n 'headers': {},
\n 'resource_usage': random.uniform(0.2, 0.8)
\n }<\/p>\n
# \u5904\u7406\u8bf7\u6c42
\n processed_request = middleware.process_request(request)<\/p>\n
# \u6a21\u62df\u54cd\u5e94
\n response = {
\n 'status_code': 200 if i % 20 != 0 else 503,
\n 'headers': {},
\n 'body': {'data': f'response_{i}'},
\n 'response_time': random.uniform(50, 2000),
\n 'from_cache': i % 3 == 0
\n }<\/p>\n
# \u5904\u7406\u54cd\u5e94
\n adapted_response = middleware.process_response(processed_request, response)<\/p>\n
# \u6bcf20\u4e2a\u8bf7\u6c42\u663e\u793a\u62a5\u544a
\n if i % 20 == 0 and i > 0:
\n report = middleware.get_adaptation_report()
\n print(f"\\\\n=== Adaptation Report after {i} requests ===")
\n print(f"Learning efficiency: {report['adaptation_efficiency']:.2%}")<\/p>\n
for rec in report['recommended_adjustments'][:2]:
\n print(f" {rec['priority']}: {rec['description']}")<\/p>\n
\u72b6\u6001\u7801\u6027\u80fd\u4f18\u5316\u662f\u4e00\u4e2a\u591a\u5c42\u6b21\u3001\u591a\u7ef4\u5ea6\u7684\u7cfb\u7edf\u5de5\u7a0b：<\/p>\n
\u6838\u5fc3\u4f18\u5316\u9886\u57df：<\/p>\n
\u54cd\u5e94\u65f6\u95f4\u4f18\u5316\u00a0– \u5206\u89e3\u548c\u5206\u6790\u5404\u4e2a\u7ec4\u4ef6\u7684\u65f6\u95f4\u6d88\u8017<\/p>\n<\/li>\n
\u7f13\u5b58\u7b56\u7565\u4f18\u5316\u00a0– \u667a\u80fd\u7684\u7f13\u5b58\u51b3\u7b56\u548c\u7ba1\u7406<\/p>\n<\/li>\n
\u4f20\u8f93\u4f18\u5316\u00a0– HTTP\/2\/3\u5934\u90e8\u538b\u7f29，CDN\u4f18\u5316<\/p>\n<\/li>\n
\u81ea\u9002\u5e94\u5904\u7406\u00a0– \u57fa\u4e8e\u4e0a\u4e0b\u6587\u7684\u72b6\u6001\u7801\u54cd\u5e94\u8c03\u6574<\/p>\n<\/li>\n
\u5173\u952e\u6280\u672f：<\/p>\n
\u6027\u80fd\u5256\u6790\u00a0– \u8be6\u7ec6\u7684\u65f6\u95f4\u5206\u89e3\u548c\u74f6\u9888\u8bc6\u522b<\/p>\n<\/li>\n
\u667a\u80fd\u7f13\u5b58\u00a0– \u57fa\u4e8e\u72b6\u6001\u7801\u548c\u4e0a\u4e0b\u6587\u7684\u52a8\u6001\u7f13\u5b58\u7b56\u7565<\/p>\n<\/li>\n
\u534f\u8bae\u4f18\u5316\u00a0– \u5145\u5206\u5229\u7528HTTP\/2\/3\u7684\u7279\u6027<\/p>\n<\/li>\n
\u673a\u5668\u5b66\u4e60\u00a0– \u81ea\u9002\u5e94\u7b56\u7565\u5b66\u4e60\u548c\u4f18\u5316<\/p>\n<\/li>\n
\u6700\u4f73\u5b9e\u8df5：<\/p>\n
\u5206\u5c42\u4f18\u5316\u00a0– \u4ece\u8fb9\u7f18\u5230\u540e\u7aef\u7684\u5168\u65b9\u4f4d\u4f18\u5316<\/p>\n<\/li>\n
\u6570\u636e\u9a71\u52a8\u00a0– \u57fa\u4e8e\u76d1\u63a7\u6570\u636e\u7684\u4f18\u5316\u51b3\u7b56<\/p>\n<\/li>\n
\u6e10\u8fdb\u5f0f\u6539\u8fdb\u00a0– \u5c0f\u6b65\u5feb\u8dd1，\u6301\u7eed\u4f18\u5316<\/p>\n<\/li>\n
\u81ea\u52a8\u5316\u8c03\u6574\u00a0– \u57fa\u4e8e\u89c4\u5219\u7684\u81ea\u52a8\u4f18\u5316<\/p>\n<\/li>\n
\u672a\u6765\u8d8b\u52bf：<\/p>\n
AI\u9a71\u52a8\u7684\u4f18\u5316\u00a0– \u66f4\u667a\u80fd\u7684\u81ea\u9002\u5e94\u7cfb\u7edf<\/p>\n<\/li>\n
\u8fb9\u7f18\u8ba1\u7b97\u00a0– \u66f4\u63a5\u8fd1\u7528\u6237\u7684\u5904\u7406<\/p>\n<\/li>\n
\u5b9e\u65f6\u4f18\u5316\u00a0– \u57fa\u4e8e\u5b9e\u65f6\u6570\u636e\u7684\u5373\u65f6\u8c03\u6574<\/p>\n<\/li>\n
\u9884\u6d4b\u6027\u4f18\u5316\u00a0– \u63d0\u524d\u9884\u6d4b\u548c\u9884\u9632\u6027\u80fd\u95ee\u9898<\/p>\n<\/li>\n
\u901a\u8fc7\u7efc\u5408\u5e94\u7528\u8fd9\u4e9b\u4f18\u5316\u7b56\u7565，\u53ef\u4ee5\u663e\u8457\u63d0\u5347\u7cfb\u7edf\u7684\u6027\u80fd\u3001\u53ef\u9760\u6027\u548c\u7528\u6237\u4f53\u9a8c\u3002<\/p>\n
\u7b2c39\u7ae0：\u72b6\u6001\u7801\u7684\u6027\u80fd\u4f18\u531639.1 \u72b6\u6001\u7801\u54cd\u5e94\u6027\u80fd\u5206\u679039.1.1 \u54cd\u5e94\u65f6\u95f4\u5206\u89e3python# \u72b6\u6001\u7801\u54cd\u5e94\u65f6\u95f4\u5206\u6790\u5de5\u5177
\nfrom typing import Dict, List, Tuple
\nimport time
\nfrom dataclasses import dataclass
\nfrom enum import Enum
\nimport statisticsclass ResponseTimeComponent(Enum):\\”\\”\\”\u54cd\u5e94\u65f6\u95f4\u7ec4\u4ef6\\”\\”\\”NETWO<\/p>\n","protected":false},"author":2,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[151,100,43],"topic":[],"class_list":["post-60023","post","type-post","status-publish","format-standard","hentry","category-server","tag-http","tag-100","tag-43"],"yoast_head":"\n