Python asyncio事件循环机制与协程调度实战详解

asyncio事件循环核心架构与运行原理

Python asyncio是原生异步编程框架，核心由事件循环（Event Loop）、协程（Coroutine）、Future/Task三层抽象构成。事件循环是整个异步体系的调度中枢——它持续监听注册的I/O事件（网络请求就绪、定时器到期等），事件触发后唤醒对应的协程继续执行。asyncio.run()封装了事件循环的创建、运行和关闭全流程，3.10+版本推荐使用，不再手动get_event_loop()。

协程函数用async def定义，调用后返回协程对象而非直接执行。协程对象需要被事件循环调度才能运行——直接调用协程函数不会有任何输出，这是asyncio初学者最常见的错误。await关键字挂起当前协程让出控制权，事件循环切换到其他可运行协程，直到await的目标完成后恢复执行。

Task调度与并发控制

asyncio.create_task()将协程包装为Task并立即提交事件循环调度，协程开始并发执行。asyncio.gather()并发运行多个协程并等待全部完成，返回结果列表。asyncio.as_completed()按完成顺序逐个获取结果，适合最快结果优先处理场景。

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def concurrent_fetch(urls, max_concurrency=10):
    semaphore = asyncio.Semaphore(max_concurrency)

    async def fetch_with_limit(session, url):
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [
            asyncio.create_task(fetch_with_limit(session, url))
            for url in urls
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

urls = [f"https://httpbin.org/get?id={i}" for i in range(50)]
results = asyncio.run(concurrent_fetch(urls, max_concurrency=5))

Semaphore是控制并发数的标准手段。不限制并发数时，asyncio.gather()会同时创建所有Task，大量并发连接可能导致对方服务器限流或本机端口耗尽。Semaphore(10)限制同时执行的协程不超过10个，其余在Semaphore内部队列排队等待。

超时控制与取消机制

asyncio.wait_for()为协程设置超时，超时后自动取消Task并抛出asyncio.TimeoutError。Task.cancel()手动取消协程，在协程下一个await点抛出CancelledError。取消机制的正确处理需要区分正常完成和被取消两种退出路径：

import asyncio

async def long_running_task():
    try:
        while True:
            await asyncio.sleep(1)
            print("工作中...")
    except asyncio.CancelledError:
        print("任务被取消，执行清理")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(3.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("主函数感知到任务已取消")

asyncio.run(main())

asyncio.timeout()是3.11+新增的异步上下文管理器，比wait_for()更灵活——可以在async with块内执行多个操作共享同一超时限制：

async def multi_step_with_timeout():
    async with asyncio.timeout(5.0):
        result1 = await step1()
        result2 = await step2()
        result3 = await step3()

异步上下文管理器与迭代器

数据库连接池、HTTP会话等资源需要正确关闭，async with异步上下文管理器确保资源释放。自定义异步上下文管理器实现__aenter__和__aexit__两个方法：

class AsyncDBPool:
    async def __aenter__(self):
        self.pool = await create_pool(host='localhost', port=5432)
        return self.pool

    async def __aexit__(self, exc_type, exc, tb):
        await self.pool.close()

async with AsyncDBPool() as pool:
    async with pool.acquire() as conn:
        result = await conn.fetch("SELECT * FROM users")

异步迭代器用__aiter__和__anext__实现，配合async for遍历。aiohttp的流式响应、asyncpg的查询结果都返回异步迭代器，逐条产出数据避免一次性加载大结果集到内存。

多线程与多进程的混合调度

asyncio事件循环运行在单线程中，CPU密集型计算仍会阻塞。asyncio.to_thread()将同步阻塞函数放到线程池执行（3.9+），asyncio.get_running_loop().run_in_executor()指定自定义线程池或进程池：

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n):
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    data = await asyncio.to_thread(open, "large_file.bin", "rb")
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = await asyncio.gather(
            loop.run_in_executor(pool, cpu_heavy, 10**7),
            loop.run_in_executor(pool, cpu_heavy, 10**7),
            loop.run_in_executor(pool, cpu_heavy, 10**7),
        )
    print(results)

asyncio.run(main())

IO密集型用asyncio原生异步，CPU密集型用ProcessPoolExecutor并行，阻塞式IO用to_thread桥接——这三种模式组合覆盖了Python异步编程的所有典型场景。