Python异步编程实战：asyncio协程任务调度与并发性能调优指南

Python的异步编程能力经过多年演进已经足够成熟，asyncio成为I/O密集型服务的主流方案。本文从事件循环、协程调度、任务并发三个层面，讲解asyncio的实际使用方式，并给出连接池、限流、超时处理等生产环境必备的调优配置。

asyncio事件循环与协程调度机制

事件循环是asyncio的核心，它维护一个就绪队列，轮询I/O事件并在就绪时恢复对应的协程。理解这个机制就能明白为什么asyncio适合I/O密集而非CPU密集任务：CPU计算不释放控制权，会阻塞其他协程。协程通过await挂起，把控制权交还给事件循环，这是协作式并发的关键。

import asyncio

async def fetch_data(url):
    # 模拟IO等待
    await asyncio.sleep(0.5)
    return f"data from {url}"

async def main():
    tasks = [fetch_data(f"url-{i}") for i in range(5)]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

asyncio.gather并发执行多个协程，所有协程完成后返回结果列表，是并发调度的最常用接口。

基于FastAPI的异步接口开发实战

FastAPI天然支持async def路由，配合数据库异步驱动可以最大化并发能力。一个典型的异步查询接口如下：

from fastapi import FastAPI
import asyncpg

app = FastAPI()
pool = None

@app.on_event("startup")
async def startup():
    global pool
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost/db",
        min_size=5, max_size=20
    )

@app.get("/users/{uid}")
async def get_user(uid: int):
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, name FROM users WHERE id = $1", uid
        )
    return {"id": row["id"], "name": row["name"]}

连接池是异步数据库访问的必备组件，min_size与max_size按并发峰值调整，过小会排队，过大会占用数据库连接资源。

协程任务调度与并发控制技巧

并发场景要避免无限创建任务。asyncio.Semaphore可以限制同时执行的协程数量：

sem = asyncio.Semaphore(20)

async def limited_fetch(url):
    async with sem:
        return await fetch_data(url)

async def main():
    urls = [...]  # 100个URL
    tasks = [limited_fetch(u) for u in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

return_exceptions=True让单个任务异常不影响其他任务结果，比默认的立即抛出异常更稳健。超时控制用asyncio.wait_for包装：

try:
    result = await asyncio.wait_for(fetch_data(url), timeout=5)
except asyncio.TimeoutError:
    result = None

异步任务调度与错误处理最佳实践

生产环境需要把任务调度独立出来。asyncio.create_task创建后台任务，配合cancel与异常捕获：

async def background_job():
    try:
        while True:
            await do_work()
            await asyncio.sleep(60)
    except asyncio.CancelledError:
        # 清理资源
        raise

task = asyncio.create_task(background_job())
# 关闭时取消任务
task.cancel()

还有三个容易踩坑的细节：gather的return_exceptions参数要按需设置；asyncio.run是唯一的事件循环入口，不要自行调用loop.run_until_complete；异步代码中严禁使用阻塞的time.sleep，必须用asyncio.sleep。遵循这些规则，Python异步服务的并发能力可以稳定支撑每秒数千请求的I/O场景。