Python 异步编程实战别让事件循环卡死你的服务一、为什么异步代码写起来简单跑起来却像阻塞一样卡死Python 的asyncio是异步编程的标配但新手最容易踩的坑就是在async函数里调用了同步阻塞操作比如requests.get、time.sleep、同步文件读写。这会导致整个事件循环被卡住所有协程都排不上队。更隐蔽的坑是“饥饿”即使没有显式阻塞如果一个协程运行时间过长它会独占事件循环其他协程根本插不上手。举个真实的例子一个异步 Web 服务同时处理 100 个请求其中 1 个请求触发了 CPU 密集计算比如图像处理耗时 5 秒。结果这 100 个请求全卡了 5 秒——因为 CPU 密集计算阻塞了事件循环其他协程无法被调度。从“异步”变“阻塞”往往只差一个忘记run_in_executor的调用。二、asyncio 到底是怎么调度的asyncio 的核心是事件循环Event Loop它维护一个就绪队列不断从中取出协程执行。当协程遇到 I/O 操作时注册回调并让出控制权当 I/O 完成时回调将协程重新加入就绪队列。这个模型在 I/O 密集场景下效率极高但在 CPU 密集场景下反而更差——因为事件循环是单线程的。flowchart TB A[事件循环] -- B[就绪队列] B -- C[取出协程执行] C -- D{协程状态} D --|I/O 等待| E[注册回调, 让出控制权] D --|计算完成| F[返回结果] D --|CPU 密集| G[阻塞事件循环!] E -- H[I/O 完成] H -- B F -- I[继续执行后续逻辑] G -- J[所有协程被阻塞] subgraph 正确模式 K[CPU 密集任务 → run_in_executor] L[阻塞 I/O → 替换为异步库] M[长时间协程 → 定期 await asyncio.sleep(0)] end K -- N[线程池执行, 不阻塞事件循环] L -- N M -- N2.1 调度粒度await是关键事件循环的调度粒度是“协程切换点”——即await表达式。两个await之间的代码是原子执行的不会被其他协程打断。这意味着如果一段代码中没有await它将独占事件循环直到执行完毕。2.2 结构化并发别留“孤儿任务”结构化并发Structured Concurrency的核心原则是所有子任务的生命周期必须包含在父任务的生命周期内。父任务创建子任务后必须等待所有子任务完成或取消后才能退出。这避免了“孤儿任务”——子任务在父任务退出后仍在运行导致资源泄漏或状态不一致。Python 3.11 引入的TaskGroup是结构化并发的标准实现使用async with创建任务组任务组退出时自动等待所有子任务完成。任何子任务抛出异常时其他子任务会被自动取消。2.3 取消与超时协程的取消通过Task.cancel()实现它向目标协程抛出CancelledError。协程可以在try/finally中捕获该异常并执行清理逻辑。超时通过asyncio.wait_for()实现超时后自动取消目标协程。三、代码实战怎么写出靠谱的异步代码3.1 异步 HTTP 客户端连接池 超时 重试import asyncio import aiohttp from typing import Optional from dataclasses import dataclass import time dataclass class HttpResponse: HTTP 响应封装 status: int body: str elapsed_ms: float class AsyncHttpClient: 异步 HTTP 客户端连接池 超时 重试 def __init__(self, max_connections: int 100, request_timeout: float 30.0, max_retries: int 2): self.max_connections max_connections self.request_timeout request_timeout self.max_retries max_retries self._session: Optional[aiohttp.ClientSession] None async def _get_session(self) - aiohttp.ClientSession: 懒初始化 ClientSession必须在事件循环内创建 if self._session is None or self._session.closed: connector aiohttp.TCPConnector( limitself.max_connections, # 启用 DNS 缓存减少重复解析开销 use_dns_cacheTrue, # 保持连接复用减少 TCP 握手 enable_cleanup_closedTrue, ) timeout aiohttp.ClientTimeout(totalself.request_timeout) self._session aiohttp.ClientSession( connectorconnector, timeouttimeout ) return self._session async def get(self, url: str, **kwargs) - HttpResponse: GET 请求带自动重试 last_error None for attempt in range(self.max_retries 1): start time.monotonic() try: session await self._get_session() async with session.get(url, **kwargs) as resp: body await resp.text() elapsed (time.monotonic() - start) * 1000 return HttpResponse( statusresp.status, bodybody, elapsed_mselapsed, ) except (aiohttp.ClientError, asyncio.TimeoutError) as e: last_error e # 指数退避重试 if attempt self.max_retries: wait_time 0.5 * (2 ** attempt) await asyncio.sleep(wait_time) raise RuntimeError( f请求失败重试 {self.max_retries} 次: {last_error} ) async def close(self): 关闭连接池 if self._session and not self._session.closed: await self._session.close()3.2 结构化并发与 TaskGroupimport asyncio from typing import Any async def fetch_multiple(urls: list[str]) - list[HttpResponse]: 并发请求多个 URL使用 TaskGroup 实现结构化并发 client AsyncHttpClient() results: dict[str, HttpResponse] {} async with asyncio.TaskGroup() as tg: async def fetch_one(url: str): try: resp await client.get(url) results[url] resp except Exception as e: # 记录失败不中断其他任务 results[url] HttpResponse( status0, bodystr(e), elapsed_ms0 ) for url in urls: tg.create_task(fetch_one(url)) await client.close() return [results[url] for url in urls] async def fetch_with_timeout(urls: list[str], timeout_per_url: float 5.0) - list[HttpResponse]: 带超时的并发请求 client AsyncHttpClient() results: dict[str, HttpResponse] {} async with asyncio.TaskGroup() as tg: async def fetch_one(url: str): try: # 为每个请求设置独立超时 resp await asyncio.wait_for( client.get(url), timeouttimeout_per_url, ) results[url] resp except asyncio.TimeoutError: results[url] HttpResponse( status0, body请求超时, elapsed_mstimeout_per_url * 1000 ) except Exception as e: results[url] HttpResponse( status0, bodystr(e), elapsed_ms0 ) for url in urls: tg.create_task(fetch_one(url)) await client.close() return [results[url] for url in urls]3.3 CPU 密集任务的异步化import asyncio from concurrent.futures import ProcessPoolExecutor from functools import partial # CPU 密集函数在进程池中执行 def cpu_intensive_task(data: bytes, threshold: float 0.5) - dict: 模拟 CPU 密集的图像处理任务 import hashlib # 在实际场景中这里可能是 NumPy 计算、PIL 图像处理等 result hashlib.sha256(data).hexdigest() return {hash: result, size: len(data)} async def process_with_executor(data: bytes) - dict: 将 CPU 密集任务提交到进程池执行 避免阻塞事件循环 loop asyncio.get_running_loop() # 使用 ProcessPoolExecutor而非 ThreadPoolExecutor # 因为 Python GIL 限制线程池无法利用多核 # 进程池可以真正并行执行 CPU 密集任务 with ProcessPoolExecutor(max_workers4) as executor: # run_in_executor 将同步函数包装为协程 result await loop.run_in_executor( executor, partial(cpu_intensive_task, data), ) return result async def batch_process(items: list[bytes]) - list[dict]: 批量处理 CPU 密集任务 loop asyncio.get_running_loop() with ProcessPoolExecutor(max_workers4) as executor: # 并发提交所有任务 futures [ loop.run_in_executor( executor, partial(cpu_intensive_task, item), ) for item in items ] # 等待所有任务完成 results await asyncio.gather(*futures) return results3.4 异步上下文管理器与信号量import asyncio class AsyncRateLimiter: 异步速率限制器控制并发请求数 def __init__(self, max_concurrent: int 10, rate_limit: float 100.0): self.semaphore asyncio.Semaphore(max_concurrent) self.rate_limit rate_limit self._tokens rate_limit self._last_refill asyncio.get_event_loop().time() async def acquire(self): 获取一个令牌 await self._refill_tokens() # 等待令牌可用 while self._tokens 1: await asyncio.sleep(0.01) await self._refill_tokens() self._tokens - 1 await self.semaphore.acquire() def release(self): 释放令牌 self.semaphore.release() async def _refill_tokens(self): 按速率补充令牌 now asyncio.get_event_loop().time() elapsed now - self._last_refill self._tokens min( self.rate_limit, self._tokens elapsed * self.rate_limit, ) self._last_refill now async def __aenter__(self): await self.acquire() return self async def __aexit__(self, *args): self.release()四、架构权衡协程、线程还是进程维度asyncio协程threading线程multiprocessing进程并发类型协作式需 await 让出抢占式OS 调度独立进程GIL 影响受限单线程受限GIL不受限内存开销极低协程 ~2KB中线程 ~8MB高进程 ~30MB适用场景I/O 密集I/O 密集 简单并行CPU 密集调试难度中堆栈追踪复杂高竞态条件低独立进程权衡一协程与线程的选择。I/O 密集场景用协程内存开销低、无锁问题CPU 密集场景用进程绕过 GIL。混合场景用协程 run_in_executor桥接。权衡二TaskGroup 与 gather。asyncio.gather是非结构化的——子任务的异常不会自动取消其他任务。TaskGroup是结构化的——任何子任务异常会取消所有其他子任务。建议新代码统一使用 TaskGroup。权衡三信号量与连接池。信号量控制并发数但不管理连接生命周期连接池既控制并发又复用连接。对于 HTTP 请求连接池aiohttp.TCPConnector比信号量更合适。五、总结异步编程的核心思路Python 异步编程的核心思路是I/O 让出、计算隔离、结构化并发。I/O 操作用await让出控制权CPU 密集任务用run_in_executor隔离到进程池子任务用TaskGroup纳入结构化生命周期——三者协同才能写出真正高效的异步代码。落地步骤第一步将所有同步 I/O 替换为异步库requests → aiohttptime.sleep → asyncio.sleep第二步用TaskGroup替换gather确保子任务的生命周期可控第三步对 CPU 密集任务使用ProcessPoolExecutor避免阻塞事件循环。关键原则是——异步编程不是把所有函数都加上async而是让 I/O 等待时 CPU 不闲着。