很多人第一次学Python异步，卡住的不是语法本身，而是脑子里的运行模型没有搭起来

async、await看起来并不复杂，但真正开始写代码时，问题很快就会冒出来，最常见的有两种：一种是代码明明写成了异步，跑起来却还是串行；另一种是程序确实并发起来了，但一碰到阻塞函数、超时控制、任务取消这些场景，整个执行流程立刻开始混乱

这篇文章不准备空讲概念，而是想把asyncio当成Python里的任务调度中枢来理解，它真正负责的不是把代码写得更花，而是让程序在等待网络、数据库、磁盘这些I/O返回的过程中，不至于傻等，而是能够把这段时间利用起来，先去处理别的任务

如果你之前总是分不清async、协程、任务、事件循环到底谁管谁，那么这篇文章就是专门用来把这条链路理顺的，上篇先把基础心智模型和核心语法讲明白，让你知道异步代码到底是怎么跑起来的；下篇再继续讲进阶能力、实战写法和常见坑，尽量让你看完之后，不只是知道有这个东西，而是真的敢上手写
 

一、asyncio到底解决什么问题

1.1 同步代码到底卡在哪

先看最普通的同步写法：

import time


def fetch_data(name, delay):
    print(f"{name} 开始请求")
    time.sleep(delay)
    print(f"{name} 请求结束")


fetch_data("任务A", 2)
fetch_data("任务B", 2)
fetch_data("任务C", 2)

这段代码的问题不在于它错，而在于它太传统了，任务A没执行完，任务B不会开始；任务B没结束，任务C也只能继续等，三个任务各等2秒，总耗时就是6秒

如果这6秒里程序一直在做大量CPU计算，那还说得过去；但现实里很多场景不是忙，而是等

典型场景包括：

- 请求第三方接口，等响应返回

- 查询数据库，等结果出来

- 读写文件，等磁盘操作完成

- 爬虫抓网页，等网络返回内容

这时候程序最浪费的不是算力，而是等待时间，asyncio的价值就在这里，它让程序在等待A的时候，可以把时间片切给B和C

1.2 asyncio 适合什么场景

asyncio最适合的是I/O密集型任务，不是CPU密集型任务

可以这样理解：

如果任务的大部分时间花在等，异步很适合

如果任务的大部分时间花在算，异步帮助不大

适合异步的场景：

Web服务接口调用

- 聊天机器人请求模型接口

- 批量爬虫抓取页面

- 异步数据库访问

- 高并发网络服务

不太适合直接用异步硬上的场景：

- 大量图像处理

- 视频编解码

- 大规模数值计算

- 纯 CPU 密集型算法

所以先记住一句话：asyncio不是加速一切的魔法，它主要解决等待期间别浪费的问题

二、asyncio四个核心概念

2.1 协程是什么

协程就是用async def定义出来的函数

例如：

async def say_hello():

    return "hello"

注意，这里调用say_hello()时，不会立刻执行函数体，而是先得到一个协程对象

coroutine_obj = say_hello()

print(coroutine_obj)

只有当它被事件循环调度，或者被await等待时，它才会真正运行

所以协程本质上不是已经执行的任务，而是一个可以被调度执行的异步函数

2.2 await是什么

await的作用不是单纯“调用函数”，而是告诉当前协程：

这个地方要等一个异步结果，在等待期间你先把控制权交出去，让事件循环去调度别的任务

最常见的例子：

import asyncio

async def main():
    print("开始")
    await asyncio.sleep(1)
    print("结束")

asyncio.run(main())

这里的asyncio.sleep(1)不会像time.sleep(1)那样把整个线程卡死，它会把当前协程挂起，让事件循环去做别的事情

2.3 事件循环是什么

事件循环是asyncio的中控台，也是整个异步系统的大脑

它主要做三件事：

1. 接收要执行的协程和任务

2. 监听谁已经准备好继续运行

3. 把 CPU 执行机会分配给合适的任务

你平时最常见到它的地方就是：

asyncio.run(main())

这句代码会帮你创建事件循环、运行协程、结束后关闭循环，对于大多数日常开发这就够用了

2.4 Task是什么

如果说协程是待执行的工作单，那Task就是已经挂到调度系统里的正式任务

看一个例子：

import asyncio

async def worker(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 结束")

async def main():
    task1 = asyncio.create_task(worker("任务A", 2))
    task2 = asyncio.create_task(worker("任务B", 1))

    await task1
    await task2

asyncio.run(main())

create_task()的意义是把协程立刻交给事件循环调度，而不是等你写到await那一行才开始执行

这个差别非常关键，因为很多我明明写了异步为什么还是串行的问题，根因就在这里

三、基础语法

3.1 async和await的最小闭环

先看一个最小可运行版本：

import asyncio

async def fetch_user():
    await asyncio.sleep(1)
    return {"name": "Tom", "age": 18}

async def main():
    user = await fetch_user()
    print(user)

asyncio.run(main())

这一套结构里最重要的是：

- async def定义协程函数

- await等待异步结果

- asyncio.run()启动整个异步入口

如果你刚入门，先把这三个东西练熟，后面的内容基本都能串起来

3.2 create_task：让多个任务真正并发起来

很多人会写出下面这种代码：

import asyncio

async def work(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 完成")

async def main():
    await work("A", 2)
    await work("B", 2)
    await work("C", 2)

asyncio.run(main())

这不是并发，这是异步串行，因为你每次都在等前一个任务跑完

真正的并发写法是：

import asyncio

async def work(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 完成")
    return name

async def main():
    task1 = asyncio.create_task(work("A", 2))
    task2 = asyncio.create_task(work("B", 2))
    task3 = asyncio.create_task(work("C", 2))

    result1 = await task1
    result2 = await task2
    result3 = await task3
    print(result1, result2, result3)

asyncio.run(main())

这样三个任务会先一起挂到事件循环里，再分别等待结果，总耗时接近2秒，而不是6秒

3.3 gather：批量收任务结果

如果你有一组任务要一起执行，asyncio.gather()会更顺手

import asyncio

async def fetch_data(i):
    await asyncio.sleep(1)
    return f"结果{i}"

async def main():
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3),
    )
    print(results)

asyncio.run(main())

它的好处是代码更紧凑，适合发起一批请求，统一收结果这种场景

但这里也有一个要点：gather()默认其中一个任务抛异常，整个等待过程会被异常打断，这个行为在项目里要提前想清楚

3.4 asyncio.sleep和time.sleep的区别

这是入门异步最容易踩的坑之一

错误示例：

import asyncio
import time

async def bad_task():
    print("开始阻塞")
    time.sleep(3)
    print("结束阻塞")

asyncio.run(bad_task())

这里虽然函数写成了async def，但只要你在协程里用了time.sleep()，整个线程还是会被卡住

正确写法：

import asyncio

async def good_task():
    print("开始等待")
    await asyncio.sleep(3)
    print("结束等待")

asyncio.run(good_task())

简单说：

- time.sleep()是真阻塞

- asyncio.sleep()是异步挂起

这个差别决定了你的程序到底是不是表面异步，实际卡死

这一篇先把基础打牢，我们重点解决asyncio到底是什么、为什么我写了异步却没有并发这两个最核心的问题