Python asyncio事件循环机制与协程任务调度原理深度解析

asyncio事件循环Event Loop核心调度机制

asyncio是Python 3.4引入的异步IO标准库，3.5增加了async/await语法支持。事件循环（Event Loop）是asyncio的核心，负责调度协程、管理IO回调、处理定时器和信号。理解Event Loop的工作原理是编写高性能异步代码的基础。

Event Loop本质是一个不断运行的循环，每次迭代做三件事：执行所有就绪的回调（包括定时器到期、IO完成通知）；等待IO事件（通过epoll/kqueue/IOCP）；处理延迟回调。在Linux上，asyncio默认使用epoll作为IO多路复用后端，Windows使用IOCP。

import asyncio

# 查看事件循环底层实现
loop = asyncio.new_event_loop()
print(type(loop._selector))
# Linux: <class 'selectors.EpollSelector'>
# macOS: <class 'selectors.KqueueSelector'>
# Windows: <class 'selectors._SelectSelector'>

# Event Loop核心循环的简化逻辑
def run_loop_step(loop):
    # 1. 计算超时时间（取最近的定时器）
    timeout = loop._schedule_timers()

    # 2. 等待IO事件
    events = loop._selector.select(timeout)

    # 3. 处理就绪的IO回调
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)

    # 4. 处理到期的定时器
    loop._run_ready_callbacks()

asyncio.run()是Python 3.7+提供的便捷入口，内部创建新事件循环、运行协程、关闭循环。在生产环境中，通常使用uvloop替代默认事件循环，uvloop基于libuv用Cython实现，性能比默认实现快2-4倍。

import asyncio
import uvloop

# 使用uvloop替代默认事件循环
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

async def main():
    await asyncio.sleep(1)
    print("Hello from uvloop")

asyncio.run(main())

async/await协程语法与任务编排模式

async def定义的函数返回coroutine对象，不是普通函数返回值。coroutine需要通过await执行或被事件循环调度。await只能在async函数内部使用，它会挂起当前协程，将控制权交回事件循环，直到被await的Future完成。

import asyncio

async def fetch_data(url, delay):
    print(f"开始请求 {url}")
    await asyncio.sleep(delay)  # 模拟IO等待
    print(f"完成请求 {url}")
    return {"url": url, "data": f"response_{delay}"}

async def main():
    # 串行执行：总耗时 = 1+2+3 = 6秒
    result1 = await fetch_data("api/users", 1)
    result2 = await fetch_data("api/posts", 2)
    result3 = await fetch_data("api/comments", 3)

    # 并发执行：总耗时 = max(1,2,3) = 3秒
    results = await asyncio.gather(
        fetch_data("api/users", 1),
        fetch_data("api/posts", 2),
        fetch_data("api/comments", 3),
    )

    # TaskGroup (Python 3.11+)：更安全的并发，任一任务异常会取消所有任务
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("api/users", 1))
        task2 = tg.create_task(fetch_data("api/posts", 2))
    # task1.result(), task2.result() 可获取结果

asyncio.run(main())

asyncio.gather和TaskGroup的核心区别在于错误处理。gather默认在任一协程抛出异常时返回异常对象（return_exceptions=True时），TaskGroup在任一任务失败时立即取消所有其他任务，保证资源清理。生产环境推荐使用TaskGroup，它提供了结构化并发的保证。

asyncio线程池与进程池混合并发模式

asyncio是单线程并发模型，CPU密集型任务会阻塞事件循环。asyncio提供了run_in_executor方法，将阻塞操作委托给线程池或进程池执行，保持事件循环的响应性。

import asyncio
import concurrent.futures
import time

def cpu_bound_task(n):
    """CPU密集型计算"""
    count = 0
    for i in range(n):
        count += i * i
    return count

def blocking_io_task():
    """阻塞IO操作（如同步数据库驱动）"""
    time.sleep(2)
    return "io_done"

async def main():
    loop = asyncio.get_running_loop()

    # 线程池：适合IO密集型阻塞操作
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
        io_result = await loop.run_in_executor(pool, blocking_io_task)

    # 进程池：适合CPU密集型计算
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as pool:
        cpu_result = await loop.run_in_executor(
            pool, cpu_bound_task, 10_000_000
        )

    print(f"IO: {io_result}, CPU: {cpu_result}")

asyncio.run(main())

选择线程池还是进程池取决于任务类型。IO阻塞操作（如requests库的同步HTTP请求、同步数据库驱动）用线程池即可，因为GIL在IO等待时释放。CPU计算密集型任务必须用进程池，因为GIL阻止多线程并行执行Python字节码。进程池的代价是进程创建和IPC开销，适合长时运行的计算任务。

aiohttp异步HTTP客户端与服务器实战

aiohttp是asyncio生态中最常用的HTTP客户端和服务器框架。客户端支持连接池复用、cookie管理、session上下文，服务器支持中间件、路由和静态文件服务。

import asyncio
import aiohttp

async def fetch_batch(urls, max_concurrent=10):
    """并发批量请求，限制最大并发数"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_one(session, url):
        async with semaphore:
            async with session.get(url) as response:
                status = response.status
                body = await response.text()
                return {"url": url, "status": status, "length": len(body)}

    connector = aiohttp.TCPConnector(
        limit=100,              # 总连接数上限
        limit_per_host=20,      # 单host连接数上限
        ttl_dns_cache=300,      # DNS缓存时间(秒)
        enable_cleanup_closed=True,
    )

    timeout = aiohttp.ClientTimeout(
        total=30,       # 总超时
        connect=10,     # 连接超时
        sock_read=10,   # 读取超时
    )

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        results = await asyncio.gather(
            *[fetch_one(session, url) for url in urls],
            return_exceptions=True,
        )

    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            print(f"请求失败 {url}: {result}")
        else:
            print(f"请求成功 {url}: {result['status']}")

    return results

async def main():
    urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(30)]
    await fetch_batch(urls, max_concurrent=10)

asyncio.run(main())

Semaphore控制最大并发数是异步爬虫和API聚合的常见模式。不限制并发会导致瞬间发起大量连接，触发服务端限流或本地端口耗尽。TCPConnector的连接池配置确保连接复用，避免频繁TCP握手开销。aiohttp的ClientSession应在应用生命周期内复用，而不是每次请求创建新session，否则连接池缓存完全失效。

# aiohttp服务器
from aiohttp import web

async def health_check(request):
    return web.json_response({"status": "ok"})

async def get_user(request):
    user_id = request.match_info["id"]
    await asyncio.sleep(0.01)
    return web.json_response({"id": user_id, "name": f"user_{user_id}"})

app = web.Application()
app.router.add_get("/health", health_check)
app.router.add_get("/users/{id}", get_user)

web.run_app(app, host="0.0.0.0", port=8080)

异步代码调试的一个常见陷阱是忘记await协程。Python 3.8+会在RuntimeWarning中提示”coroutine was never awaited”。使用asyncio.all_tasks()可以查看事件循环中所有活跃任务，排查协程泄漏问题。在asyncio项目中集成APM监控时，需要确保监控SDK支持asyncio上下文传播，否则trace_id会在await切换时丢失。