asyncio 是 Python 标准库中用于编写单线程并发代码的强大工具。它非常适合处理网络请求、文件读写、数据库查询等 I/O 密集型任务。

你可以把 asyncio 想象成一个极其高效的“总调度员”（事件循环）。当程序遇到需要等待的操作（比如等待网页加载）时，它不会傻傻地干等，而是立刻转去处理其他任务。等等待的操作完成后，再回来继续处理。

下面是一份从入门到实战的 asyncio 用法教程：

🎯 核心概念与基础语法

在 asyncio 中，有三个最重要的关键词：

• async def：用来定义一个协程（Coroutine）。你可以把它理解为一个“可以暂停和恢复的函数”。
• await：只能在协程内部使用，用来“等待”一个耗时的异步操作完成。遇到 await 时，程序会主动把控制权交还给调度员，去执行别的任务。
• asyncio.run()：程序的入口，用来启动并运行最外层的协程。

基础代码模板：

import asyncio

# 1. 使用 async def 定义协程
async def say_hello():
    print("你好！")
    # 2. 使用 await 等待一个异步操作（这里模拟等待1秒）
    # 注意：千万不要用 time.sleep()，它会卡死整个程序！
    await asyncio.sleep(1) 
    print("世界！")

# 3. 使用 asyncio.run() 启动协程
asyncio.run(say_hello())

🚀 进阶用法：实现并发执行

如果直接 await 多个协程，它们依然是按顺序（串行）执行的。想要让它们同时（并发）跑起来，我们需要使用任务（Task）。

1. 使用 asyncio.gather() 批量并发
这是最常用的并发方式，适合把一堆任务一起提交，并等待它们全部完成。

import asyncio

async def fetch_data(id, delay):
    print(f"任务 {id} 开始获取数据...")
    await asyncio.sleep(delay)  # 模拟网络请求耗时
    print(f"任务 {id} 获取数据完成！")
    return f"任务 {id} 的结果"

async def main():
    # 将三个任务并发执行，gather 会等待它们全部结束
    results = await asyncio.gather(
        fetch_data(1, 2),
        fetch_data(2, 1),
        fetch_data(3, 3)
    )
    print("所有任务结束，拿到结果：", results)

asyncio.run(main())
# 你会发现任务1、2、3几乎是同时开始打印“开始获取数据”的，总耗时约3秒而不是6秒。

2. 使用 asyncio.create_task() 创建后台任务
如果你希望某个任务在后台悄悄运行，而主程序继续做别的事情，可以使用它。

import asyncio

async def background_task():
    print("后台任务：开始持续监控...")
    await asyncio.sleep(5)
    print("后台任务：监控结束！")

async def main():
    # 创建后台任务，它会立刻被调度执行，但不会阻塞下面的代码
    task = asyncio.create_task(background_task())
    
    print("主程序：我先去处理点别的业务...")
    await asyncio.sleep(2)
    print("主程序：业务处理完毕，等待后台任务收尾...")
    
    # 等待后台任务彻底完成
    await task

asyncio.run(main())

💡 实战避坑指南

在使用 asyncio 时，新手非常容易踩以下几个坑，请务必注意：

1. 严禁在协程中使用同步阻塞代码
这是最大的禁忌。如果你在 async def 里使用了 time.sleep()、requests.get() 或普通的文件读写，整个事件循环都会被卡死，所有并发都会失效。

• 解决办法：必须使用异步库（如用 aiohttp 代替 requests，用 aiofiles 代替普通文件操作）。
• 如果非要用同步代码：可以使用 await asyncio.to_thread(同步函数) 把它扔到独立的线程池里去执行，避免阻塞主线程。

2. 只有 await 才能真正触发执行
如果你定义了一个协程但没有 await 它，Python 只会创建一个协程对象，而不会真正执行里面的代码。

• 错误写法：coro = say_hello() （没有任何输出）
• 正确写法：await say_hello() 或 asyncio.run(say_hello())

3. 记得设置超时控制
网络请求等外部操作很容易无限挂起。使用 asyncio.wait_for() 可以设置超时时间，防止程序卡死。

try:
    # 等待 fetch_data 任务，最多等 2 秒，超时就报错
    result = await asyncio.wait_for(fetch_data(1, 10), timeout=2.0)
except asyncio.TimeoutError:
    print("任务超时了！")

4. 结构化并发（Python 3.11+ 新特性）
如果你使用的是 Python 3.11 及以上版本，强烈推荐使用 asyncio.TaskGroup()。它可以确保一组任务要么全部成功，要么只要有一个失败，其他任务就会被自动取消，极大地简化了错误处理。

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data(1, 1))
        task2 = tg.create_task(fetch_data(2, 2))
    # 退出 with 块时，保证所有任务都已成功完成
    print(task1.result(), task2.result())

掌握这些基础与进阶用法，基本上就能应对 Python 中绝大多数的异步编程场景了。建议从简单的 asyncio.run() 和 asyncio.gather() 开始动手实践！