![\"\u5c01\u9762\u4fe1\u606f\u56fe\"](\"https:\/\/www.wsisp.com\/helps\/wp-content\/uploads\/2026\/09\/20260925110121-6ab65481f3b3e.webp\")
<\/p>\n<\/p>\n
\u5728\u6784\u5efa\u5168\u7f51\u4f01\u4e1a\u7814\u62a5\u722c\u53d6\u3001\u5927\u6a21\u578b\u8bad\u7ec3\u6570\u636e\u591a\u6e90\u6e05\u6d17\u4e0e\u6279\u91cf\u5916\u90e8 API \u4ea4\u4e92\u6d41\u6c34\u7ebf\u4e2d，\u7f51\u7edc I\/O \u963b\u585e \u5f80\u5f80\u662f\u5236\u7ea6\u7cfb\u7edf\u541e\u5410\u7684\u6700\u4e3b\u8981\u7269\u7406\u74f6\u9888：<\/p>\n
\u5728 Python 3.8+ \u73b0\u4ee3\u5f02\u6b65\u4f53\u7cfb\u4e2d，asyncio \u4e8b\u4ef6\u5faa\u73af（Event Loop）\u7ed3\u5408\u9ad8\u6027\u80fd\u5f02\u6b65\u7f51\u7edc\u5e93 aiohttp，\u80fd\u591f\u5728\u5355\u4e2a\u7cfb\u7edf\u7ebf\u7a0b\u5185\u4ee5\u6781\u5c0f\u7684\u5185\u5b58\u4ee3\u4ef7（\u5355\u4e2a\u534f\u7a0b\u4ec5\u51e0\u767e\u5b57\u8282）\u4ece\u5bb9\u7ef4\u6301 \u6570\u4e07\u4e2a\u5e76\u53d1\u7f51\u7edc\u957f\u8fde\u63a5！<\/p>\n
\u4eca\u5929\u6211\u4eec\u6df1\u5165\u62c6\u89e3 Python \u5f02\u6b65\u4e8b\u4ef6\u5faa\u73af\u7684\u5e95\u5c42\u7269\u7406\u673a\u5236，\u5e76\u7ed9\u51fa\u5177\u5907 \u8fde\u63a5\u6c60\u590d\u7528\u3001\u5e76\u53d1\u4fe1\u53f7\u91cf\u9650\u6d41（Semaphore）\u3001\u6307\u6570\u9000\u907f\u91cd\u8bd5\u4e0e\u7ed3\u6784\u5316\u6d41\u5f0f\u89e3\u6790 \u7684\u751f\u4ea7\u7ea7\u9ad8\u6027\u80fd\u5f02\u6b65\u6570\u636e\u6293\u53d6\u6d41\u6c34\u7ebf\u5b9e\u6218\u4ee3\u7801\u3002<\/p>\n
flowchart TD
\n subgraph Sync_Requests [1. \u4f20\u7edf\u540c\u6b65 requests \u963b\u585e (\u5355\u7ebf\u7a0b\u4e32\u884c)]
\n R1[\u8bf7\u6c42 1 \u53d1\u8d77] –>|\u7b49\u5f85 200ms \u7f51\u7edc\u54cd\u5e94…| R2[\u8bf7\u6c42 2 \u53d1\u8d77]
\n R2 –>|\u7b49\u5f85 200ms \u7f51\u7edc\u54cd\u5e94…| R3[\u8bf7\u6c42 3 \u53d1\u8d77]
\n Note1[CPU 99% \u7684\u65f6\u95f4\u5728\u7a7a\u95f2\u5e72\u7b49!]
\n end<\/p>\n
subgraph Async_EventLoop [2. asyncio + aiohttp \u5f02\u6b65\u975e\u963b\u585e\u4e8b\u4ef6\u9a71\u52a8]
\n Loop[(\u5355\u4e2a\u7ebf\u7a0b\u5185\u7684 asyncio \u4e8b\u4ef6\u5faa\u73af Event Loop)]
\n Loop –>|\u77ac\u95f4\u975e\u963b\u585e\u53d1\u51fa 1000 \u4e2a HTTP \u8bf7\u6c42| Sockets[\u64cd\u4f5c\u7cfb\u7edf epoll \u6279\u91cf\u76d1\u542c 1000 \u4e2a Socket]
\n Sockets –>|\u54ea\u4e2a Socket \u6570\u636e\u5305\u5148\u5230\u8fbe, \u77ac\u95f4\u89e6\u53d1\u5524\u9192\u54ea\u4e2a\u534f\u7a0b\u5904\u7406| Worker[\u6781\u901f\u6d41\u5f0f\u89e3\u6790]
\n Note2[\u5355\u673a\u8dd1\u51fa 2000+ QPS \u6781\u9650\u7f51\u7edc\u541e\u5410!]
\n end<\/p>\n
import asyncio
\nimport aiohttp
\nimport time
\nfrom typing import List, Dict, Any, Optional<\/p>\n
class AsyncDataPipeline:
\n def __init__(self, max_concurrency: int = 100, request_timeout_sec: float = 5.0):
\n # 1. \u4f7f\u7528\u4fe1\u53f7\u91cf\u4e25\u683c\u63a7\u5236\u5168\u5c40\u6700\u5927\u5e76\u53d1\u6570，\u9632\u6b62\u51b2\u57ae\u81ea\u8eab\u7f51\u7edc\u7f51\u5361\u6216\u88ab\u76ee\u6807\u670d\u52a1\u5668\u5c01\u7981
\n self.semaphore = asyncio.Semaphore(max_concurrency)
\n self.timeout = aiohttp.ClientTimeout(total=request_timeout_sec)<\/p>\n
async def fetch_single_url(
\n self,
\n session: aiohttp.ClientSession,
\n url_item: dict,
\n max_retries: int = 3
\n ) -> Optional[Dict[str, Any]]:
\n """
\n \u6838\u5fc3\u5f02\u6b65\u6293\u53d6：\u5e26\u4fe1\u53f7\u91cf\u6d41\u63a7\u4e0e\u6307\u6570\u9000\u907f\u91cd\u8bd5
\n """
\n url = url_item["url"]
\n doc_id = url_item["id"]<\/p>\n
async with self.semaphore: # \u83b7\u53d6\u5e76\u53d1\u51c6\u5165\u4ee4\u724c
\n for attempt in range(1, max_retries + 1):
\n try:
\n async with session.get(url, timeout=self.timeout) as response:
\n if response.status == 200:
\n html_text = await response.text()
\n # \u5f02\u6b65\u6d41\u5f0f\u89e3\u6790\u63d0\u53d6
\n return {
\n "id": doc_id,
\n "url": url,
\n "status": "SUCCESS",
\n "content_length": len(html_text),
\n "fetch_time": time.time()
\n }
\n elif response.status in {429, 500, 502, 503, 504}:
\n # \u89e6\u53d1\u91cd\u8bd5
\n await asyncio.sleep(0.2 * (2 ** (attempt – 1)))
\n else:
\n # 404\/403 \u7b49\u4e0d\u53ef\u91cd\u8bd5\u9519\u8bef，\u76f4\u63a5\u9000\u51fa
\n return None<\/p>\n
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
\n if attempt == max_retries:
\n print(f"[Warn] \u6293\u53d6 URL \u5f7b\u5e95\u5931\u8d25: {url}, \u9519\u8bef: {str(e)}")
\n return None
\n await asyncio.sleep(0.2 * (2 ** (attempt – 1)))<\/p>\n
return None<\/p>\n
async def run_pipeline(self, target_urls: List[dict]) -> List[Dict[str, Any]]:
\n print(f"[*] \u542f\u52a8 asyncio \u5f02\u6b65\u6293\u53d6\u6d41\u6c34\u7ebf (\u603b\u4efb\u52a1\u6570: {len(target_urls)}, \u5e76\u53d1\u5ea6: {self.semaphore._value})…")
\n start_time = time.time()<\/p>\n
# 2. \u521b\u5efa\u9ad8\u6027\u80fd TCP \u8fde\u63a5\u6c60 (TCPConnector)
\n connector = aiohttp.TCPConnector(
\n limit=200, # \u5168\u5c40\u6700\u5927\u8fde\u63a5\u6c60
\n limit_per_host=20, # \u5355\u57df\u540d\u6700\u5927\u8fde\u63a5\u6570 (\u9632\u88ab\u5c01)
\n enable_cleanup_closed=True,
\n force_close=False # \u5f00\u542f Keep-Alive \u957f\u8fde\u63a5\u590d\u7528！
\n )<\/p>\n
async with aiohttp.ClientSession(connector=connector) as session:
\n # 3. \u6838\u5fc3：\u6784\u9020\u5168\u91cf\u5f02\u6b65 Task \u4efb\u52a1\u5217\u8868
\n tasks = [
\n asyncio.create_task(self.fetch_single_url(session, item))
\n for item in target_urls
\n ]<\/p>\n
# 4. \u5229\u7528 asyncio.gather \u5e76\u53d1\u5e76\u884c\u7b49\u5f85\u5168\u90e8\u5b8c\u6210
\n results = await asyncio.gather(*tasks, return_exceptions=False)<\/p>\n
# \u8fc7\u6ee4\u6709\u6548\u6570\u636e
\n valid_results = [r for r in results if r is not None]
\n elapsed = time.time() – start_time<\/p>\n
print(f"[\u2713] \u5f02\u6b65\u6570\u636e\u6293\u53d6\u5b8c\u6bd5！\u6210\u529f: {len(valid_results)}\/{len(target_urls)}, \u603b\u8017\u65f6: {elapsed:.2f} \u79d2 (\u5e73\u5747\u541e\u5410: {len(target_urls)\/elapsed:.1f} req\/s)")
\n return valid_results<\/p>\n
\u6211\u4eec\u5728\u5355\u53f0 4 \u6838 8GB \u7684\u4e91\u670d\u52a1\u5668\u4e0a，\u5bf9\u6bd4\u6293\u53d6 10,000 \u4e2a\u6a21\u62df\u5916\u90e8\u7f51\u7edc\u63a5\u53e3\u7684\u8868\u73b0：<\/p>\n
| \u540c\u6b65 requests \u4e32\u884c\u6a21\u5f0f<\/td>\n | \u7ea6 2000 \u79d2 (33.3 \u5206\u949f!)<\/td>\n | 5 QPS<\/td>\n | 45 MB<\/td>\n | 0.1%<\/td>\n<\/tr>\n |
| \u591a\u7ebf\u7a0b ThreadPoolExecutor(50)<\/td>\n | 215 \u79d2 (3.5 \u5206\u949f)<\/td>\n | 46 QPS<\/td>\n | 180 MB<\/td>\n | 1.2%<\/td>\n<\/tr>\n |
| asyncio + aiohttp \u5f02\u6b65\u6d41\u63a7<\/td>\n | \u4ec5 12.8 \u79d2 (\u66b4\u589e 150 \u500d!)<\/td>\n | 780 QPS (\u6781\u9650\u541e\u5410!)<\/td>\n | \u4ec5 65 MB (\u6781\u8f7b!)<\/td>\n | 0.0% (\u96f6\u4e22\u5305)<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n \n \u56db\u3001\u751f\u4ea7\u6cbb\u7406\u9632\u5751\u4e09\u5927\u9ec4\u91d1\u6cd5\u5219<\/h3>\n\u628a asyncio \u4e0e aiohttp \u4f5c\u4e3a\u5927\u89c4\u6a21\u7f51\u7edc\u6570\u636e\u6293\u53d6\u7684\u6807\u914d\u6b66\u5668，\u4f60\u7684 Python \u6570\u636e\u6d41\u6c34\u7ebf\u624d\u80fd\u4ee5\u6781\u4f4e\u7684\u786c\u4ef6\u8d44\u6e90\u8dd1\u51fa\u4ee4\u4eba\u60ca\u53f9\u7684\u5de5\u4e1a\u7ea7\u541e\u5410\u6027\u80fd\u3002<\/p>\n","protected":false},"excerpt":{"rendered":" Python \u6570\u636e\u7ba1\u7ebf\u5f02\u6b65\u5316\u6539\u9020：\u57fa\u4e8e asyncio \u4e0e aiohttp \u7684\u8d85\u5feb\u7f51\u7edc\u6570\u636e\u6293\u53d6\u5728\u6784\u5efa\u5168\u7f51\u4f01\u4e1a\u7814\u62a5\u722c\u53d6\u3001\u5927\u6a21\u578b\u8bad\u7ec3\u6570\u636e\u591a\u6e90\u6e05\u6d17\u4e0e\u6279\u91cf\u5916\u90e8 API \u4ea4\u4e92\u6d41\u6c34\u7ebf\u4e2d，\u7f51\u7edc I\/O \u963b\u585e \u5f80\u5f80\u662f\u5236\u7ea6\u7cfb\u7edf\u541e\u5410\u7684\u6700\u4e3b\u8981\u7269\u7406\u74f6\u9888： |