1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
| class ImprovedAPIService:
"""改进后的API服务"""
def __init__(self):
self.base_urls = {
"service_a": "https://httpbin.org",
"service_b": "https://jsonplaceholder.typicode.com",
"service_c": "https://httpbin.org"
}
self.session = None
self._session_lock = asyncio.Lock()
self.connector_config = {
"limit": 100,
"limit_per_host": 30,
"ttl_dns_cache": 300,
"use_dns_cache": True,
"keepalive_timeout": 60,
"enable_cleanup_closed": True
}
async def get_session(self) -> aiohttp.ClientSession:
"""获取全局session（懒加载）"""
if self.session is None or self.session.closed:
async with self._session_lock:
if self.session is None or self.session.closed:
connector = aiohttp.TCPConnector(**self.connector_config)
self.session = aiohttp.ClientSession(
connector=connector,
timeout=aiohttp.ClientTimeout(total=10.0)
)
print("创建全局session和优化的连接池")
return self.session
async def fetch_data_improved(self, request_data: dict) -> dict:
"""改进版数据获取"""
session = await self.get_session()
results = {}
tasks = []
for service_name, base_url in self.base_urls.items():
task = asyncio.create_task(
self.fetch_from_service_improved(session, service_name, base_url, request_data)
)
tasks.append((service_name, task))
for service_name, task in tasks:
try:
result = await task
results[service_name] = result
except Exception as e:
results[service_name] = {"error": str(e)}
return results
async def fetch_from_service_improved(self, session: aiohttp.ClientSession,
service_name: str, base_url: str,
request_data: dict) -> dict:
"""改进版服务请求"""
endpoints = {
"service_a": "/delay/1",
"service_b": "/posts/1",
"service_c": "/get"
}
url = f"{base_url}{endpoints.get(service_name, '/get')}"
try:
start_time = time.time()
async with session.get(url, params=request_data) as response:
end_time = time.time()
if response.status == 200:
content = await response.text()
return {
"status": "success",
"response_time": end_time - start_time,
"content_length": len(content),
"connection_reused": True
}
else:
return {"status": "error", "code": response.status}
except Exception as e:
return {"status": "error", "message": str(e)}
async def close(self):
"""优雅关闭"""
if self.session and not self.session.closed:
await self.session.close()
print("全局session已关闭")
class ImprovedAPIAggregator:
"""改进版API聚合器"""
def __init__(self):
self.api_service = ImprovedAPIService()
self.connection_monitor = ConnectionMonitor()
async def handle_request_improved(self, request_data: dict) -> dict:
"""改进版请求处理"""
start_time = time.time()
result = await self.api_service.fetch_data_improved(request_data)
end_time = time.time()
result["total_response_time"] = end_time - start_time
return result
async def handle_batch_requests_improved(self, requests: List[dict]) -> List[dict]:
"""改进版批量请求处理"""
tasks = [
asyncio.create_task(self.handle_request_improved(req))
for req in requests
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
async def close(self):
"""关闭资源"""
await self.api_service.close()
async def performance_comparison_test():
"""性能对比测试"""
print("=== 性能对比测试开始 ===")
concurrent_requests = 20
request_data = {"test": "performance"}
print("\n--- 测试改进版本 ---")
aggregator = ImprovedAPIAggregator()
start_time = time.time()
requests = [{"batch_id": i, **request_data} for i in range(concurrent_requests)]
results = await aggregator.handle_batch_requests_improved(requests)
end_time = time.time()
success_count = sum(1 for r in results if isinstance(r, dict) and
r.get("total_response_time") is not None)
avg_response_time = sum(r.get("total_response_time", 0) for r in results
if isinstance(r, dict)) / len(results)
print(f"改进版结果:")
print(f" 总耗时: {end_time - start_time:.2f}s")
print(f" 成功率: {success_count}/{concurrent_requests}")
print(f" 平均响应时间: {avg_response_time:.3f}s")
await aggregator.close()
|
