要理解asyncio，得先接受一个反直觉的事实：异步并不等于多线程。一个标准的asyncio程序通常只运行在一个线程里，却能同时处理成百上千个网络连接。它依赖的不是操作系统级别的线程切换，而是事件循环不断检查哪些协程可以继续执行。协程用async def定义，内部通过await挂起当前逻辑，把控制权交还给事件循环。等待期间，事件循环会切换到其他准备好的协程，这样IO耗时就被大量并发任务分摊了。
真正开始使用asyncio时，首先需要明确两个层级：定义协程和运行协程。定义协程只是创建一个协程对象，并不会执行任何代码；只有将它交给事件循环后，函数体才会被驱动。最常用的入口是asyncio.run(main())，它会创建新的事件循环、运行主协程、最后清理资源。事件循环内部维护一个任务队列，某个协程执行到await时，会返回一个挂起点，事件循环随即调度其他任务。这也是asyncio适合IO密集型场景的原因：CPU并没有空转，只是在IO等待期间去执行别的代码了。
一、事件循环与协程的基本运行机制
协程是asyncio中最基础的执行单元。使用async def声明的函数被称为协程函数，调用它不会直接运行函数体，而是返回一个协程对象。比如fetch_data()调用后如果不去await它，会得到一个尚未开始的协程对象，并可能触发运行时警告。只有通过await、asyncio.run()或asyncio.create_task()等方式将协程交给事件循环，代码才会真正执行。
事件循环可以理解为一个单线程调度器。它不断询问所有已注册的任务，哪个任务已经准备好继续运行，哪个任务还在等待IO。某个协程执行到await asyncio.sleep(1)时，它会向事件循环注册一个1秒后的回调，然后立即返回。事件循环不会傻等，而是切换去执行其他协程。这就是为什么协程必须在可能阻塞的位置使用await，如果你调用time.sleep(1)而不是await asyncio.sleep(1)，整个线程会被阻塞，事件循环无法调度其他任务，异步优势也就消失了。
import asyncio
async def task(name, delay):
print(f"{name} 开始，等待 {delay} 秒")
await asyncio.sleep(delay)
print(f"{name} 结束")
return name
async def main():
# gather 会同时驱动多个协程
result_a, result_b = await asyncio.gather(
task("A", 2),
task("B", 1),
)
print(f"结果: {result_a}, {result_b}")
asyncio.run(main())上面的代码中，两个协程会在同一时间窗口内等待，而不是串行等待2秒再等1秒。总耗时取决于较长的那个任务，即约2秒。这个例子也说明，异步并发并不是并行计算，它不会让CPU密集型计算变快；它的收益主要体现在数据库查询、HTTP请求、文件读取这类需要等待外部资源返回的操作上。
二、任务创建与并发结果收集
把多个协程同时提交给事件循环有三种常见方式。第一种是直接await asyncio.gather(coro1, coro2)，语法简单，适合一次性收集所有结果。第二种是使用asyncio.create_task()把协程包装成Task对象，这样协程会立即进入事件循环调度，而你可以在稍后再await它。第三种是使用asyncio.as_completed()，它返回一个迭代器，哪个任务先完成就先处理哪个任务，适合实时消费结果的场景。
create_task与直接await协程的最大区别在于启动时机。直接await coro会在当前协程中逐项等待，前一个没完成就不会启动后一个；create_task则会把协程排入事件循环任务队列，立即获得一个可等待对象，多个任务可以同时处于等待状态。下面的示例演示了先创建多个Task，再用gather统一等待结果。
import asyncio
async def download(item_id):
await asyncio.sleep(0.2)
return f"item-{item_id}"
async def main():
tasks = []
for item_id in range(10):
# 创建任务后，download 会很快进入等待状态
tasks.append(asyncio.create_task(download(item_id)))
finished, pending = await asyncio.wait(
tasks,
timeout=1.5,
return_when=asyncio.ALL_COMPLETED,
)
for task in finished:
print(await task)
for task in pending:
task.cancel()
asyncio.run(main())这里用了asyncio.wait，它比gather更灵活，可以设置总超时时间，也能通过return_when指定何时返回。需要注意，wait返回的是集合，迭代顺序不保证与任务创建顺序一致；如果需要按原始顺序取结果，应优先使用gather。还有一点经常被忽略：如果pending中有被取消的任务，结束前应当显式取消它们，避免事件循环退出时出现警告。
三、网络请求实战：用信号量控制并发
异步编程最常见的落地场景是并发网络请求。标准库中的urllib.request是阻塞的，如果直接在协程中调用它，线程会卡住，其他协程无法运行。解决办法有两个：一是把阻塞调用丢给线程池，通过loop.run_in_executor()或asyncio.to_thread()执行；二是直接使用支持异步的HTTP客户端，比如aiohttp或httpx。下面的示例使用aiohttp，它在内部基于asyncio实现，不会阻塞事件循环。
import asyncio
import aiohttp
async def fetch(session, url, semaphore):
async with semaphore:
async with session.get(url) as response:
text = await response.text()
print(f"{url} 返回 {len(text)} 字符")
return text
async def main():
urls = [f"https://ipipp.com/api/items/{i}" for i in range(20)]
semaphore = asyncio.Semaphore(5)
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url, semaphore) for url in urls]
results = await asyncio.gather(*tasks)
print(f"共完成 {len(results)} 个请求")
asyncio.run(main())如果不加限制，20个请求会同时建立连接，可能触发目标服务器的频率限制，也可能耗尽本地文件描述符。asyncio.Semaphore(5)将并发上限控制在5个，当某个请求获得信号量后才继续执行，释放前其他任务需要等待。这样既保留了异步并发能力，又避免对远程服务造成过大压力。生产环境中通常还会加入重试机制和超时控制，比如用asyncio.wait_for给单个请求包裹最大等待时间。
还需要注意aiohttp.ClientSession的复用。不要在每一个请求函数里重复创建Session，因为每次创建都涉及连接池初始化，开销较大，而且关闭不当可能导致连接泄漏。推荐在main中使用async with创建一次Session，然后把它传入各个任务函数。
四、异常处理、超时与同步阻塞陷阱
异步代码中的异常不会像同步代码那样直接向上抛出。某个子任务抛出的异常，通常会被gather收集，并在主协程中重新抛出。如果你不希望一个任务失败影响其他任务，可以在每个任务内部捕获异常，或者使用gather(..., return_exceptions=True)，让异常作为结果返回而不中断整体流程。
import asyncio
async def risky_task(item_id):
await asyncio.sleep(0.1)
if item_id % 2 == 0:
raise ValueError(f"任务 {item_id} 失败")
return item_id * 10
async def main():
tasks = [risky_task(i) for i in range(5)]
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
print(f"捕获异常: {result}")
else:
print(f"正常结果: {result}")
asyncio.run(main())超时控制同样重要。网络请求、数据库连接等操作如果没有超时机制，协程可能无限挂起，导致整个程序难以退出。asyncio.wait_for可以给单协程设置最大等待时间，超时后抛出asyncio.TimeoutError。下面的代码展示了如何用它为任务加锁时间。
import asyncio
async def slow_operation():
await asyncio.sleep(5)
return "完成"
async def main():
try:
result = await asyncio.wait_for(slow_operation(), timeout=1.0)
print(result)
except asyncio.TimeoutError:
print("操作超时，已取消")
asyncio.run(main())最后必须强调同步阻塞调用的问题。很多开发者把time.sleep、requests.get、cursor.execute直接放在协程里，结果事件循环被卡死，所有异步任务全部停止。正确做法是：能使用异步库就尽量使用异步库；如果只有阻塞版本，就通过await asyncio.to_thread(func, *args)把阻塞操作交给线程池执行。虽然线程池的并发能力远不如协程，但至少可以防止阻塞事件循环。异步编程的全部收益，都建立在事件循环始终有可调度任务这个前提上，一旦出现长时间同步阻塞，整个程序就会退化成比顺序执行还慢的串行脚本。
Python asyncio异步编程事件循环修改时间：2026-09-25 08:38:36