Python3 协程 asyncio

如果你使用 Python 写过多线程网络程序，一定遇到过这些问题：线程创建销毁开销大、GIL 锁限制 CPU 并行、线程间共享数据容易出错等问题。为了解决这些问题，推荐使用 asyncio 库，asyncio 的解决思路也很简单 —— 不用线程，用单线程 + 事件循环处理成千上万个 IO 连接。本章从 async/await 语法讲起，覆盖事件循环、Task、异步队列等知识点，最后写一个并发下载器。读完本章你应该能用 asyncio 替代 90% 的多线程 IO 场景。

为什么需要 asyncio？

线程模型在 IO 密集型任务里有两个硬伤：

1. 内存开销：每个线程默认栈 8MB 左右（Linux 系统默认线程栈，8MB（ulimit -s 8192，而 Python threading.Thread 创建的是操作系统原生线程，不改变内核栈大小，即 Python 线程大小为 8MB），一万个连接就是 8 万 MB 内存，不现实。

2. 切换开销：线程由操作系统调度，切换要保存/恢复寄存器状态，还要考虑锁竞争。高并发下大量 CPU 时间花在调度上，而不是业务上。

asyncio 的核心假设是：程序大部分时间都在等网络响应、磁盘读写、数据库查询。既然在等，干嘛不让出 CPU 去干别的？协程（coroutine）就是这么做的 —— 遇到 IO 操作就挂起自己，事件循环去调度其他协程，IO 就绪后再唤醒。所有协程跑在一个线程里，没有锁，没有上下文切换，内存占用只有协程对象本身的几百字节。

一句话：线程是抢占式多任务，协程是协作式多任务。前者靠操作系统强插，后者靠程序员主动让出。

async/await：协程的定义与运行

async def 用来定义协程函数。例如：

async def func():
    pass

直接调用 func() 不会执行函数内部代码，只会返回一个 协程对象（coroutine），而且普通函数内部不能写 await，只有 async def 定义的函数内部才能使用 await。

await 用来等待一个可等待对象（协程、Task、Future），实现协程挂起、让出事件循环。简单说，当遇到 IO 阻塞时，先暂停当前协程，去执行其他就绪协程，IO 完成后再回来继续。例如

import asyncio

async def hello():
    print("开始")
    # 模拟IO等待：此时协程让出控制权，事件循环可以调度其他任务
    await asyncio.sleep(1)
    print("结束")

# 得到协程对象，代码不运行
coro = hello()
print(coro)

# 交给事件循环执行
asyncio.run(coro)

运行示例，输出如下：

<coroutine object hello at 0x000001BE87C02C80>
开始
结束

关键点：

• async def 定义的是协程函数，调用它返回一个协程对象，不是直接执行。

• await 后面必须跟可等待对象（awaitable）：协程、Task、Future。它表示“这里可能要等，先去干别的”。

• asyncio.run() 是 Python 3.7+ 的入口函数，负责创建事件循环、运行协程、关闭循环，一个程序里通常只调用一次。

顺序执行

顺序执行，即串行执行，代码依次运行，上一个任务完全结束，才开始下一个。总耗时 = 所有任务耗时累加，例如：

import asyncio
import time

async def fetch(url: str):
    """模拟网络延迟任务，返回协程"""
    print(f"Fetching {url}...")
    await asyncio.sleep(1)  # 模拟 1 秒网络延迟
    print(f"Done {url}")

async def main_serial():
    """顺序执行：3 个 URL 需要 3 秒"""
    start = time.time()
    await fetch("url-1")
    await fetch("url-2")
    await fetch("url-3")
    print(f"Serial: {time.time() - start:.2f}s")

asyncio.run(main_serial())

运行代码，输出如下：

Fetching url-1...
Done url-1
Fetching url-2...
Done url-2
Fetching url-3...
Done url-3
Serial: 3.03s

从输出可以看出，一次执行 url-1、url-2 和 url-3，耗时 3 秒。

并发执行

并发执行指多个任务同时发起，遇到 IO 阻塞时让出线程，事件循环调度其他任务。所有任务的等待时间重叠，任务总耗时 ≈ 耗时最长的那一个任务。

可以使用 asyncio.gather() 或 asyncio.create_task() 实现并发执行。

asyncio.gather()

接收多个可等待对象（协程、Task），统一调度并发执行，等待全部任务完成，最后按传入顺序收集所有返回结果。语法如下：

asyncio.gather(*aws, return_exceptions=False)

参数说明：

• *aws：可变参数，传入一堆协程 / Task

• return_exceptions（可选）

• 默认 False：任意一个任务抛异常，整体立刻抛出异常，其余任务可能继续运行但结果丢弃

• 设置 True：异常不会向上抛出，异常实例会作为该任务对应的返回值放入结果列表

简单示例：

import asyncio
import time

async def fetch(url: str):
    """模拟网络延迟任务，返回协程"""
    print(f"Fetching {url}...")
    await asyncio.sleep(1)  # 模拟 1 秒网络延迟
    print(f"Done {url}")

async def main_concurrent():
    """并发执行：3 个 URL 只需约 1 秒"""
    start = time.time()
    await asyncio.gather(
        fetch("url-1"),
        fetch("url-2"),
        fetch("url-3")
    )
    print(f"Concurrent: {time.time() - start:.2f}s")

asyncio.run(main_concurrent())

运行示例，输出如下：

Fetching url-1...
Fetching url-2...
Fetching url-3...
Done url-1
Done url-2
Done url-3
Concurrent: 1.00s

asyncio.gather() 把多个协程交给事件循环同时调度。当 url-1 的 sleep 挂起时，url-2 和 url-3 已经开始执行了。三个协程几乎同时进入等待状态，1 秒后同时被唤醒，总耗时接近 1 秒而不是 3 秒。

asyncio.create_task()

用于把协程对象包装成 Task，立刻交给事件循环后台调度执行。

和 asyncio.gather() 的主要区别：

• gather() 偏向批量并发 + 统一收集结果

• create_task() 偏向手动创建后台任务，持有句柄自由控制

语法如下：

asyncio.create_task(coro, *, name=None)

参数说明：

• coro：协程对象（async def() 调用得到）

• name：可选，任务名称，方便日志调试

返回值是一个 Task 对象。

简单示例：

import asyncio
import time

async def fetch(url: str):
    """模拟网络延迟任务，返回协程"""
    print(f"Fetching {url}...")
    await asyncio.sleep(1)  # 模拟 1 秒网络延迟
    print(f"Done {url}")

async def main_concurrent():
    """并发执行：3 个 URL 只需约 1 秒"""
    start = time.time()
    # 创建任务
    task1=asyncio.create_task(fetch("url-1"))
    task2=asyncio.create_task(fetch("url-2"))
    task3=asyncio.create_task(fetch("url-3"))
    # 等待任务
    await task1
    await task2
    await task3
    print(f"Concurrent: {time.time() - start:.2f}s")

asyncio.run(main_concurrent())

运行示例，输出如下：

Fetching url-1...
Fetching url-2...
Fetching url-3...
Done url-1
Done url-2
Done url-3
Concurrent: 1.02s

如果你要获取任务的返回值，可以这样做：

import asyncio

async def hello(delay):
    """模拟耗时"""
    await asyncio.sleep(delay)
    return f"sleep {delay} done"

async def main():
    # 创建任务，此时已经开始跑了
    task1 = asyncio.create_task(hello(1))
    task2 = asyncio.create_task(hello(2))

    # 等待任务完成，获取返回值
    r1 = await task1
    r2 = await task2
    print(r1)
    print(r2)

asyncio.run(main())

运行示例，输出如下：

sleep 1 done
sleep 2 done

常见错误：在普通函数里 await

await 关键字只允许出现在用 async def 定义的函数内部。

普通 def 函数的代码上下文不支持挂起、恢复协程，Python 语法层面直接禁止。例如：

async def coro():
    await asyncio.sleep(1)

# 错误！普通函数里不能用 await
def normal():
    # await coro()  # SyntaxError
    pass

# 正确：把普通函数也改成 async def，或者用 asyncio.run() 启动入口
async def main():
    await coro()

# 或者在普通函数里创建事件循环（不推荐）
def normal_workaround():
    asyncio.run(coro())

注意了，asyncio 是传染性的，一旦某个函数里用了 await，调用链上层的所有函数通常都应该是 async def。这也是很多人从同步代码迁移到 asyncio 时最痛苦的地方 —— 需要重构整条调用链。

在普通方法中调用 asyncio.run() 后，普通方法不会变为 async def 方法，这是因为 async def 是语法标记，必须显式写在函数头部。函数内部调用什么代码，也不能改变函数自身类型。而且，asyncio.run() 只是一个普通同步函数。

Python广告

事件循环：asyncio 的心脏

事件循环就是一个无限循环的调度器，不停检查「有没有事情要做」，有事就执行，没事就等待事件到来。

下面以一家奶茶店来通俗介绍时间循环，假设奶茶店只有 1 个店员（单线程）

• 店员 = 主线程

• 事件循环 = 店员的工作流程

• 协程任务 = 订单

• IO 等待（await sleep / 网络请求）= 等待奶茶泡好

流程：

• 处理订单 A，需要等奶茶浸泡（await）

• 不用原地傻等，先搁置订单 A，去处理订单 B

• 等奶茶泡好（IO 就绪），再回头继续做完订单 A

这个循环轮流处理就绪任务的工作模式，就是事件循环。Python asyncio 的所有协程，全靠它驱动。

事件循环（Event Loop）负责两件事：

（1）维护一个可等待对象的队列

（2）监听 IO 就绪事件（通过操作系统的 select / epoll / kqueue），唤醒对应的协程

asyncio.run 背后发生了什么

前面已经使用 asyncio.run() 启动了协程：

asyncio.run(main())

上面语句等价于下面的手动操作：

import asyncio

async def main():
    """定义顶层协程函数"""
    print("Running in loop")
    # 让出事件循环，休眠0.5秒（非阻塞等待）
    await asyncio.sleep(0.5)
    print("Done")

# 手动创建全新的事件循环对象
loop = asyncio.new_event_loop()
# 将新建的循环设置为当前线程的默认事件循环
asyncio.set_event_loop(loop)
try:
    # 运行协程，直到协程执行完毕
    loop.run_until_complete(main())
finally:
    # 关闭所有未完成的异步生成器，释放资源
    loop.run_until_complete(loop.shutdown_asyncgens())
    # 关闭事件循环，回收资源
    loop.close()

运行示例，输出如下：

Running in loop
Done

asyncio.run() 做了三件事：

1. 调用 asyncio.new_event_loop() 创建新循环；

2. 调用 run_until_complete() 运行主协程；

3. 主协程结束后，清理并关闭循环。

获取当前事件循环

在协程内部用 asyncio.get_running_loop() 获取当前事件循环。例如：

import asyncio

async def check_loop():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()
    print(f"Current loop: {loop}")
    print(f"Is running: {loop.is_running()}")

asyncio.run(check_loop())

运行示例，输出如下：

Current loop: <ProactorEventLoop running=True closed=False debug=False>
Is running: True

注意：get_event_loop() 在 Python 3.10+ 已经弃用，因为它在不存在循环时会自动创建隐式循环，容易让人踩坑。

在事件循环里调度回调

有时候你需要把同步代码的回调塞进事件循环，可以这样做：

import asyncio

async def main():
    # 获取当前正在运行的事件循环实例
    loop = asyncio.get_running_loop()
    
    # 普通同步回调函数（不能使用await）
    def sync_callback():
        print("This runs in the event loop thread")
    
    # 安排回调：延迟1秒后在事件循环中执行同步函数
    loop.call_later(1, sync_callback)
    
    # 安排回调：尽快执行（不会立刻抢占当前代码，要等当前协程主动让出控制权）
    loop.call_soon(sync_callback)
    
    # 休眠2秒，给回调足够时间触发执行
    await asyncio.sleep(2)

asyncio.run(main())

运行示例，输入如下：

This runs in the event loop thread
This runs in the event loop thread

注意，call_soon 和 call_later 是事件循环的底层调度接口，框架开发者用得更多。日常业务代码直接用 asyncio.create_task 和 await 即可。

Task：把协程丢进事件循环并发跑

Task 是对协程进行封装的对象，它会将协程注册到事件循环当中，交由事件循环完成调度与执行工作。如果把协程比作一份 “等待被执行的计划”，那么 Task 就相当于一份 “已经正式提交给调度器、等待调度运行的任务”。

create_task：立即启动后台任务

asyncio.create_task () 主要作用是将协程自动封装为可调度的任务，并尽快交由事件循环并发执行，以此实现多个异步任务的并行运行，区别于直接 await 协程的串行执行方式。

调用该函数后会返回 Task 对象，开发者可以借助这个对象完成任务等待、获取执行结果、捕获异常、主动取消任务等操作。

注意：创建任务时协程不会立刻执行，最终由事件循环统一调度运行，同时使用过程中也需要关注任务生命周期与异常处理相关细节。

import asyncio

async def worker(name: str, delay: float):
    print(f"[{name}] Start")
    await asyncio.sleep(delay)
    print(f"[{name}] Done after {delay}s")

async def main():
    # create_task 立即把协程注册到事件循环，不会阻塞当前协程
    task1 = asyncio.create_task(worker("A", 2.0))
    task2 = asyncio.create_task(worker("B", 1.0))
    
    print("Tasks created, doing other things...")
    await asyncio.sleep(0.5)
    print("Now waiting for tasks...")
    
    # await task 会等待它完成，并返回结果
    await task1
    await task2
    print("All done")

asyncio.run(main())

运行示例，输出如下：

Tasks created, doing other things...
[A] Start
[B] Start
Now waiting for tasks...
[B] Done after 1.0s
[A] Done after 2.0s
All done

gather：等待多个任务完成

asyncio.gather() 能够接收多个协程对象，将它们统一安排并发执行，待全部任务运行完成后，按照传入协程的先后顺序汇总并返回所有任务结果。例如：

import asyncio

async def fetch(url: str) -> str:
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    # gather 返回结果列表，顺序与传入顺序一致
    results = await asyncio.gather(
        fetch("https://api.a.com"),
        fetch("https://api.b.com"),
        fetch("https://api.c.com")
    )
    for r in results:
        print(r)

asyncio.run(main())

运行代码，输出如下：

Data from https://api.a.com
Data from https://api.b.com
Data from https://api.c.com

gather 的返回值是列表，顺序保证。如果某个任务抛异常，默认行为是立即传播异常，其他任务继续运行。可以用 return_exceptions=True 改为把异常对象放进结果列表：

results = await asyncio.gather(
    fetch("ok"),
    fetch("bad"),  # 假设这里抛异常
    return_exceptions=True
)
# results[1] 将是异常对象，不是抛出

wait_for：给任务加超时

asyncio.wait_for() 主要用于为异步任务设置最长执行时限，等待指定协程运行完成。若协程在设定超时时间内顺利执行完毕，函数会正常返回协程执行结果；一旦超出规定时限任务仍未结束，就会抛出 asyncio.TimeoutError 超时异常，同时主动终止对应的协程运行，能够有效避免异步任务无限阻塞，方便开发者管控异步程序的运行时长。例如：

import asyncio

async def slow_task():
    # 休眠10秒
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        # 等待协程完成，最多等待2秒
        result = await asyncio.wait_for(slow_task(), timeout=2.0)
        print(result)
    except asyncio.TimeoutError:
        print("Task timed out!")

asyncio.run(main())

运行代码，输出如下：

Task timed out!

注意，wait_for() 在超时后会取消任务。如果协程内部没有处理 CancelledError，它会收到异常并退出。如果你希望超时后任务继续在后台跑（不取消），需要用 asyncio.wait + timeout 参数配合 return_when。

return_when 三种可选枚举值：

• asyncio.FIRST_COMPLETED：任意一个任务完成 / 失败，立即返回

• asyncio.FIRST_EXCEPTION：出现第一个异常才返回，全部成功则等待全部结束

• asyncio.ALL_COMPLETED：等待所有任务完成（此时 timeout 只是最长总等待时长）

例如：

import asyncio

async def long_task():
    print("长任务开始执行")
    await asyncio.sleep(5)
    print("长任务执行完毕（即使外层等待超时依然会跑完）")

async def main():
    # 创建任务，立刻调度进入循环
    task = asyncio.create_task(long_task())

    # 最多等待2秒，超时不取消任务
    done, pending = await asyncio.wait(
        [task],
        timeout=2,
        return_when=asyncio.FIRST_COMPLETED
    )

    if task in pending:
        print("等待超时！任务仍在后台继续运行，不取消")

    # 如果主函数提前退出，事件循环关闭会杀死后台任务；
    # 若需要等待后台任务最终完成，可以使用 await 等待任务完成
    await task

asyncio.run(main())

运行代码，输出如下：

长任务开始执行
等待超时！任务仍在后台继续运行，不取消
长任务执行完毕（即使外层等待超时依然会跑完）

注意，如果不添加 await task 语句，任务会在主函数退出后结束，任务不会执行完毕。

取消任务与清理

可以使用 cancel() 手动取消一个任务，取消任务时，协程内部会在下一个 await 点抛出 CancelledError。如果协程吞掉了这个异常（没有 raise 抛出），task.cancelled() 会返回 False，调用者以为任务还在跑，这很危险。

注意了，在 except CancelledError 里做完清理后，一定要重新抛出。

例如：

import asyncio

# 定义一个长任务
async def long_task():
    try:
        while True:
            print("Working...")
            await asyncio.sleep(1)
    except asyncio.CancelledError: # 捕获取消任务异常
        print("Cleanup: saving state...")
        raise  # 必须重新抛出，让调用者知道确实取消了

async def main():
    # 创建任务
    task = asyncio.create_task(long_task())
    # 暂行2.5秒
    await asyncio.sleep(2.5)
    # 取消任务
    task.cancel()
    try:
        await task
    except asyncio.CancelledError: # 捕获取消任务异常
        print("Task was cancelled")

asyncio.run(main())

运行示例，输出如下：

Working...
Working...
Working...
Cleanup: saving state...
Task was cancelled

asyncio.Queue：协程间的异步队列

线程用 queue.Queue 做线程间通信，协程用 asyncio.Queue 做协程间通信。两者 API 几乎一样，但 asyncio.Queue 的 get() 和 put() 是协程，会挂起而不是阻塞线程。

生产者-消费者模型

使用 asyncio.Queue 实现生产者和消费者之间通信，例如：

import asyncio
import random

async def producer(queue: asyncio.Queue, name: str):
    """生产者协程：持续往队列放入数据"""
    for i in range(5):
        item = f"{name}-item-{i}"
        # 将数据放入队列，队列满时此处会阻塞等待有空位
        await queue.put(item)
        print(f"[Producer {name}] Put {item}")
        # 生成 0.1 ~ 0.5 之间随机小数，休眠，模拟生产耗时
        await asyncio.sleep(random.uniform(0.1, 0.5))

async def consumer(queue: asyncio.Queue, name: str):
    """消费者协程：循环从队列取出并处理数据"""
    while True:
        # 阻塞等待获取队列中的元素
        item = await queue.get()
        # 哨兵值 None，收到代表没有更多任务，消费者需要退出
        if item is None:
            # 标记当前任务处理完成
            queue.task_done()
            break
        print(f"[Consumer {name}] Got {item}")
        # 模拟业务处理耗时
        await asyncio.sleep(random.uniform(0.2, 0.6))
        # 当前取出的任务已经处理完毕
        queue.task_done()

async def main():
    # 创建异步有界队列，最大容量10，避免无限堆积占用内存
    queue = asyncio.Queue(maxsize=10)

    # 创建2个生产者任务
    producers = [
        asyncio.create_task(producer(queue, f"P{i}"))
        for i in range(2)
    ]
    # 创建3个消费者任务
    consumers = [
        asyncio.create_task(consumer(queue, f"C{i}"))
        for i in range(3)
    ]

    # 等待所有生产者全部生产完成
    await asyncio.gather(*producers)
    print("All producers done, sending sentinel...")

    # 向队列投放与消费者数量相等的哨兵值 None
    # 每个消费者读取到一个None就自行退出循环
    for _ in consumers:
        await queue.put(None)

    # 等待所有消费者正常退出
    await asyncio.gather(*consumers)
    print("All done")

asyncio.run(main())

运行示例，效果如下图：

[Producer P0] Put P0-item-0
[Producer P1] Put P1-item-0
[Consumer C0] Got P0-item-0
[Consumer C1] Got P1-item-0
[Producer P1] Put P1-item-1
[Consumer C2] Got P1-item-1
[Producer P0] Put P0-item-1
[Consumer C1] Got P0-item-1
[Producer P1] Put P1-item-2
[Consumer C0] Got P1-item-2
[Producer P0] Put P0-item-2
[Producer P1] Put P1-item-3
[Consumer C2] Got P0-item-2
[Consumer C0] Got P1-item-3
[Producer P0] Put P0-item-3
[Consumer C1] Got P0-item-3
[Producer P1] Put P1-item-4
[Producer P0] Put P0-item-4
[Consumer C2] Got P1-item-4
[Consumer C0] Got P0-item-4
All producers done, sending sentinel...
All done

注意：

• maxsize=10 是有界队列，满了后 put() 会挂起，直到消费者取走数据。这是背压（backpressure）机制，防止生产者过快压垮消费者。

• task_done() 通知队列　"一个任务处理完了"，配合 await queue.join() 可以等所有数据被处理完。上面示例为了可读性用了 gather()，生产代码建议 await queue.join()。

• None 作为哨兵值是一种常见做法，也可以用 asyncio.Event 通知消费者退出。

Python广告

与线程 Queue 的对比

下表从四个方面对 queue.Queue 和 asyncio.Queue 进行比较：

注意：不要在协程里用 queue.Queue！它的阻塞会阻塞整个事件循环，导致其他协程无法运行。同理，也不要在普通线程里用 asyncio.Queue。

实战：并发图片下载器

该示例把前面已经学过的知识串起来，写一个带限速、超时和错误处理的并发下载器。下载使用 aiohttp 库做异步 HTTP 请求。完整步骤如下：

（1）安装依赖

pip install aiohttp

安装日志如下：

C:\Users\Administrator>pip install aiohttp
Requirement already satisfied: aiohttp in D:\Program Files\Python313\Lib\site-packages (3.14.1)
Requirement already satisfied: aiohappyeyeballs>=2.5.0 in D:\Program Files\Python313\Lib\site-packages (from aiohttp) (2.7.1)
Requirement already satisfied: aiosignal>=1.4.0 in D:\Program Files\Python313\Lib\site-packages (from aiohttp) (1.4.0)
Requirement already satisfied: attrs>=17.3.0 in D:\Program Files\Python313\Lib\site-packages (from aiohttp) (25.3.0)
Requirement already satisfied: frozenlist>=1.1.1 in D:\Program Files\Python313\Lib\site-packages (from aiohttp) (1.8.0)
Requirement already satisfied: multidict<7.0,>=4.5 in D:\Program Files\Python313\Lib\site-packages (from aiohttp) (6.7.1)
Requirement already satisfied: propcache>=0.2.0 in D:\Program Files\Python313\Lib\site-packages (from aiohttp) (0.5.2)
Requirement already satisfied: yarl<2.0,>=1.17.0 in D:\Program Files\Python313\Lib\site-packages (from aiohttp) (1.24.5)
Requirement already satisfied: idna>=2.0 in D:\Program Files\Python313\Lib\site-packages (from yarl<2.0,>=1.17.0->aiohttp) (3.10)

[notice] A new release of pip is available: 26.0.1 -> 26.1.2
[notice] To update, run: python.exe -m pip install --upgrade pip

（2）实现下载器，代码如下：

import asyncio
import aiohttp
import os
from pathlib import Path

async def download_one(
    session: aiohttp.ClientSession,
    url: str,
    save_dir: Path,
    semaphore: asyncio.Semaphore
) -> str:
    """
    下载指定URL地址的资源，并保存到指定路径
    :param session aiohttp的会话对象，用来执行资源下载
    :param url 资源地址
    :param save_dir 下载后保存的路径
    :param semaphore 信号量，用来控制最大并发
    :return 返回提示信息
    """
    
    async with semaphore:  # 限制并发数
        try:
            # 下载资源
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                # 资源下载失败
                if resp.status != 200:
                    return f"[SKIP] {url} -> HTTP {resp.status}"
                
                # 拼接保存文件完整路径
                # os.path.basename(url)：提取url最后的文件名
                # / 是Path对象特有的路径拼接运算符，自动适配 Windows/Linux 斜杠
                filename = save_dir / os.path.basename(url)

                # 异步读取HTTP响应的二进制字节（bytes）
                # resp 是 aiohttp 请求响应对象
                # await resp.read()：下载完整二进制内容，适合图片、视频、文件
                data = await resp.read()

                # Path 对象方法：直接以二进制模式写入文件
                # 等价 open(filename, "wb").write(data)，语法更简洁
                filename.write_bytes(data)
                
                return f"[OK] {url} -> {filename} ({len(data)} bytes)"
        except asyncio.TimeoutError:
            return f"[TIMEOUT] {url}"
        except Exception as e:
            return f"[ERROR] {url} -> {e}"

async def download_all(urls: list[str], save_dir: str, max_concurrent: int = 5):
    """
    下载指定的所有URL地址资源
    :param urls 待下载的所有URL地址列表
    :param save_dir 下载后资源的保存目录
    :param max_concurrent 最大并发数
    """
    # 获取下载后资源保存路径
    save_path = Path(save_dir)
    # 创建目录
    save_path.mkdir(parents=True, exist_ok=True)
    # 创建信号量，用来控制最大并发
    semaphore = asyncio.Semaphore(max_concurrent)
    
    # 创建会话对象
    async with aiohttp.ClientSession() as session:
        # 为每个URL创建一个协程
        tasks = [
            download_one(session, url, save_path, semaphore)
            for url in urls
        ]
        # 等待所有任务完成
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for r in results:
            print(r)

if __name__ == '__main__':
    # 测试用随机图片 URL
    urls = [
        "https://picsum.photos/150",
        "https://picsum.photos/200",
        "https://picsum.photos/300",
    ] * 5  # 15 个下载任务
    
    asyncio.run(download_all(urls, "downloads", max_concurrent=5))

运行示例，输出如下：

[OK] https://picsum.photos/150 -> downloads\150 (2248 bytes)
[OK] https://picsum.photos/200 -> downloads\200 (11038 bytes)
[OK] https://picsum.photos/300 -> downloads\300 (9820 bytes)
[OK] https://picsum.photos/150 -> downloads\150 (4834 bytes)
[OK] https://picsum.photos/200 -> downloads\200 (6613 bytes)
[OK] https://picsum.photos/300 -> downloads\300 (8502 bytes)
[OK] https://picsum.photos/150 -> downloads\150 (7825 bytes)
[OK] https://picsum.photos/200 -> downloads\200 (10681 bytes)
[OK] https://picsum.photos/300 -> downloads\300 (22097 bytes)
[OK] https://picsum.photos/150 -> downloads\150 (5104 bytes)
[OK] https://picsum.photos/200 -> downloads\200 (7342 bytes)
[OK] https://picsum.photos/300 -> downloads\300 (12594 bytes)
[OK] https://picsum.photos/150 -> downloads\150 (5114 bytes)
[OK] https://picsum.photos/200 -> downloads\200 (7162 bytes)
[OK] https://picsum.photos/300 -> downloads\300 (16313 bytes)

上面例子展示了 asyncio 工程中的几个习惯用法：

1. Semaphore 限流：防止同时打开过多 HTTP 连接，保护对方服务器也保护自己内存。

2. 共享 Session：aiohttp.ClientSession() 内部维护 TCP 连接池，复用连接比每个请求新建连接快得多。

3. Timeout：ClientTimeout(total=10) 给整个请求设 10 秒上限，避免挂死。

4. 异常处理：return_exceptions=True 让 gather 不因为单个任务失败而中断，最后统一打印结果。

自己动手去试试吧！

常见陷阱与调试技巧

协程和多线程编程一样，一不小心就会踩坑，下面是一些常见的坑，看看你踩中过没有：

• 协程没执行：只定义了 async def，没 await 或 create_task，检查调用链，确保每个协程对象都被挂到事件循环上。

• 事件循环已运行：多次调用 asyncio.run() 或嵌套启动，一个程序只应有一次 asyncio.run()，子协程用 await。

• 阻塞了事件循环：在协程里调用了同步阻塞函数（如 requests.get），用 await asyncio.to_thread(sync_func) 或换成异步库。

• 任务异常被吞掉：create_task 后没有 await，异常在后台静默，给 Task 加 add_done_callback 检查 result()

• 队列死锁：消费者挂了但队列还在被 put，用 try/finally 确保消费者异常退出时发哨兵或清空队列

• 忘记 task_done()：await queue.join() 永远等不到，每个 get() 对应一个 task_done()

• 获取循环错误：用 get_event_loop() 在循环未创建时调用，用 get_running_loop() 或 asyncio.run()

调试：打印所有待运行任务

使用 asyncio.all_tasks() 获取所有任务，然后逐一输出每个任务的状态，例如：

import asyncio

async def show_tasks():
    tasks = asyncio.all_tasks()
    current = asyncio.current_task()
    for t in tasks:
        if t is current:
            continue
        print(f"Task {t.get_name()}: done={t.done()}, cancelled={t.cancelled()}")

async def monitor():
    # 监控：每隔0.3秒打印一次任务状态
    for _ in range(4):
        await show_tasks()
        print("-"*50)
        await asyncio.sleep(0.3)

async def fetch(url: str):
    print(f"Fetching {url}...")
    await asyncio.sleep(1)
    print(f"Done {url}")

async def main():
    # 启动监控任务
    monitor_task = asyncio.create_task(monitor())

    await asyncio.gather(
        fetch("url-1"),
        fetch("url-2"),
        fetch("url-3")
    )
    await monitor_task

asyncio.run(main())

把同步代码桥接到异步

同步函数阻塞了事件循环，我们可以将阻塞函数放到线程池里运行，例如：

import asyncio

def sync_blocking_func():
    import time
    time.sleep(2)  # 同步阻塞
    return "done"

async def main():
    # to_thread 把同步函数丢到线程池，不阻塞事件循环
    result = await asyncio.to_thread(sync_blocking_func)
    print(result)

asyncio.run(main())

注意，asyncio.to_thread 是 Python 3.9+ 的标准做法。如果你还在用旧版本，用 loop.run_in_executor(None, sync_func) 等价。