异步编程，就是程序在等待某件事时，不傻等，而是先去做别的事。
适合：
-
网络请求
-
下载文件
-
查询数据库
-
调用大模型 API
-
Redis、消息队列
-
WebSocket
-
同时处理大量连接
因为这些工作主要消耗 CPU，而异步主要解决“等待时间太多”的问题
1. 先理解同步编程
- 普通 Python 代码默认是同步执行的。
import time
def task(name):
print(f"{name} 开始")
time.sleep(2)
print(f"{name} 结束")
task("任务1")
task("任务2")
task("任务3")
"""
执行过程：
任务1开始
等待2秒
任务1结束
任务2开始
等待2秒
任务2结束
任务3开始
等待2秒
任务3结束
总耗时大约：
2 + 2 + 2 = 6秒
"""
这就像一个人去饭店：
效率显然不高
2. 异步编程是什么
异步的思路是：
类似：
任务1：正在等待网络返回
任务2：正在等待数据库返回
任务3：正在等待文件读取 CPU：你们都在等，那我先处理其他任务
这不是说三个任务真的同时占用三个 CPU。
而是它们在等待期间互相切换。
3. 异步编程最基本的三个关键词
async
await
asyncio
3.1 async def
普通函数：
def hello():
print("你好")
异步函数：
async def hello():
print("你好")
使用 async def 定义的函数叫：
- 协程函数
调用协程函数时，不会马上执行。
async def hello():
print("你好")
result = hello()
print(result)
"""
<coroutine object hello at 0x000001DDF55B4AC0>
sys:1: RuntimeWarning: coroutine 'hello' was never awaited
"""
报错是因为：
hello()返回的是一个“协程对象”，不是立即执行函数，需要把它交给事件循环运行。
3.2 asyncio.run()
正确执行异步函数：
import asyncio
async def hello():
print("你好")
asyncio.run(hello())
负责启动异步系统，并执行最外层的异步函数。
import asyncio
async def main():
print("程序开始")
if __name__ == "__main__":
asyncio.run(main())
3.3 await
await 的意思可以理解为：
这个任务现在要等待，先把执行权让出去，等它完成以后再回来。
import asyncio
async def task():
print("任务开始")
await asyncio.sleep(2)
print("任务结束")
asyncio.run(task())
await asyncio.sleep(2)不是单纯傻等两秒，它会告诉事件循环等待两秒，这段时间可以运行其他协程。
4. time.sleep() 和 asyncio.sleep() 的区别
这是异步编程最重要的区别之一:
time.sleep(2)：阻塞等待:全体都有，暂停两秒
asyncio.sleep()：异步等待:当前任务暂停2s，其他任务先走
5. 写了 async 不代表自动并发
先看代码：
import asyncio
async def task(name):
print(f"{name} 开始")
await asyncio.sleep(2)
print(f"{name} 结束")
async def main():
await task("任务1")
await task("任务2")
await task("任务3")
asyncio.run(main())
结果：
任务1 开始
任务1 结束
任务2 开始
任务2 结束
任务3 开始
任务3 结束
虽然函数使用了 async，但它仍然是依次执行，总耗时还是大约 6 秒。
async 只代表它是异步函数，不代表多个任务自动并发执行。
6. 真正让多个异步任务并发执行
使用：asyncio.gather()
import asyncio
import time
async def task(name):
print(f"{name} 开始")
await asyncio.sleep(2)
print(f"{name} 结束")
return f"{name} 的结果"
async def main():
start = time.time()
results = await asyncio.gather(
task("任务1"),
task("任务2"),
task("任务3"),
)
print(results)
print(f"总耗时：{time.time() – start:.2f} 秒")
asyncio.run(main())
结果：
任务1 开始
任务2 开始
任务3 开始
任务1 结束
任务2 结束
任务3 结束
['任务1 的结果', '任务2 的结果', '任务3 的结果']
总耗时：2.01 秒
三个任务都是等到2秒。
await 会让当前函数暂停，等任务完成后才往下走，
gather会同时启动 三个任务。
7. 事件循环是什么
异步程序背后有一个重要角色：Event Loop，事件循环.
可以把事件循环想象成一个调度员。
它不断检查：
- 任务1能继续运行了吗？
- 任务2的网络请求回来了吗？
- 任务3的等待结束了吗？
- 任务4的数据库结果返回了吗？
大概流程：
事件循环
↓
执行任务1
↓
任务1遇到 await，需要等待
↓
切换到任务2
↓
任务2遇到 await，需要等待
↓
切换到任务3
↓
任务1等待结束
↓
继续执行任务1
协程的切换通常发生在 await 的位置,如果一个异步函数中完全没有 await，它运行时仍然可能一直占着事件循环导致代码卡死。
8. 协程、任务、事件循环的关系
协程 Coroutine就是一个可以暂停和恢复的函数。
任务 Task是被事件循环安排执行的协程。
可以理解成：
-
协程：一份待执行的工作
-
Task：已经登记到事件循环里的工作
事件循环 Event Loop负责调度这些 Task。
协程函数
↓ 调用
协程对象
↓ create_task
Task
↓
事件循环调度执行
9. asyncio.create_task() 的作用
import asyncio
async def task(name, seconds):
print(f"{name} 开始")
await asyncio.sleep(seconds)
print(f"{name} 完成")
return name
async def main():
task1 = asyncio.create_task(task("任务1", 3))
task2 = asyncio.create_task(task("任务2", 2))
task3 = asyncio.create_task(task("任务3", 1))
print("三个任务已经启动")
result1 = await task1
result2 = await task2
result3 = await task3
print(result1, result2, result3)
asyncio.run(main())
结果：
三个任务已经启动
任务1 开始
任务2 开始
任务3 开始
任务3 完成
任务2 完成
任务1 完成
任务1 任务2 任务3
关键代码：
task1 = asyncio.create_task(…)
意思是：把协程注册为任务，让事件循环尽快调度它,而不是立即等待它完成。
10. 直接 await 和 create_task() 的区别
直接 await:
await task1()
await task2()
先等待task1完成
再执行task2
create_task:
t1 = asyncio.create_task(task1())
t2 = asyncio.create_task(task2())
await t1
await t2
先启动task1
再启动task2
然后等待它们完成
所以他们可以并发执行
11. gather() 和 create_task() 怎么选
只想同时执行多个任务
使用：
results = await asyncio.gather(
task1(),
task2(),
task3(),
)
想先启动，后面再处理
使用：
task1 = asyncio.create_task(do_something())
task2 = asyncio.create_task(do_something_else())
print("这里可以继续做其他事情")
result1 = await task1
result2 = await task2
12. 异步任务的返回值
异步函数和普通函数一样，可以使用 return
import asyncio
async def calculate():
await asyncio.sleep(1)
return 100
async def main():
result = await calculate()
print(result)
asyncio.run(main())
多个返回值：
import asyncio
async def get_user(user_id):
await asyncio.sleep(1)
return {
"id": user_id,
"name": f"用户{user_id}"
}
async def main():
results = await asyncio.gather(
get_user(1),
get_user(2),
get_user(3),
)
print(results)
asyncio.run(main())
13. 异步异常处理
异步函数同样使用 try…except
import asyncio
async def task():
await asyncio.sleep(1)
raise ValueError("任务出错了")
async def main():
try:
await task()
except ValueError as e:
print(f"捕获异常：{e}")
asyncio.run(main())
gather() 中有一个任务报错
import asyncio
async def task1():
await asyncio.sleep(1)
return "任务1成功"
async def task2():
await asyncio.sleep(1)
raise ValueError("任务2失败")
async def main():
try:
results = await asyncio.gather(
task1(),
task2(),
)
print(results)
except Exception as e:
print(f"出现异常：{e}")
asyncio.run(main())
这时异常不会直接让 gather() 中断，而是作为结果返回。
14. 设置超时时间
假设一个网络请求长时间没返回，不应该永远等下去。
可以使用：asyncio.wait_for()
import asyncio
async def request_data():
await asyncio.sleep(5)
return "请求成功"
async def main():
try:
result = await asyncio.wait_for(
request_data(),
timeout=2,
)
print(result)
except asyncio.TimeoutError:
print("请求超时")
asyncio.run(main())
因为任务需要 5 秒，而超时时间只有 2 秒，所以输出：请求超时
15. 取消异步任务
Task 可以取消:
import asyncio
async def task():
try:
print("任务开始")
await asyncio.sleep(10)
print("任务完成")
except asyncio.CancelledError:
print("任务被取消")
raise
async def main():
t = asyncio.create_task(task())
await asyncio.sleep(2)
t.cancel()
try:
await t
except asyncio.CancelledError:
print("主程序确认任务已经取消")
asyncio.run(main())
16. 限制并发数量
假设有 1000 个下载任务:await asyncio.gather(*tasks)
可能会一瞬间发出 1000 个请求,可能导致：
- 服务器拒绝连接
- 本机连接数过多
- API 限流
- 内存占用过高
- 数据库连接池被打满
这时可以使用：asyncio.Semaphore,它叫信号量，可以限制同时运行的任务数量。
import asyncio
semaphore = asyncio.Semaphore(3)
async def download(index):
async with semaphore:
print(f"任务 {index} 开始")
await asyncio.sleep(2)
print(f"任务 {index} 完成")
async def main():
tasks = [
download(i)
for i in range(10)
]
await asyncio.gather(*tasks)
asyncio.run(main())
虽然有 10 个任务，但同一时刻最多只有 3 个任务进入
17. 生产者和消费者：asyncio.Queue
异步队列适合处理流水线任务:
-
生产者：不断产生下载地址
-
消费者：从队列中取地址并下载
import asyncio
async def producer(queue):
for i in range(5):
item = f"任务{i}"
await queue.put(item)
print(f"生产：{item}")
await asyncio.sleep(0.5)
await queue.put(None)
async def consumer(queue):
while True:
item = await queue.get()
if item is None:
break
print(f"消费：{item}")
await asyncio.sleep(1)
queue.task_done()
async def main():
queue = asyncio.Queue()
await asyncio.gather(
producer(queue),
consumer(queue),
)
asyncio.run(main())
队列为空时：消费者会暂停，但不会堵住整个程序。
18. 异步适合处理的任务
- 异步适合 I/O 密集型任务
- 不适合 CPU 密集型任务
19. 普通阻塞函数怎么放进异步程序
有些老代码是同步阻塞函数，例如：
import time
def blocking_task():
time.sleep(5)
return "完成"
直接在异步函数里调用：会阻塞事件循环。
现代 Python 可以用：asyncio.to_thread()
它会把普通阻塞函数放到线程中执行，避免堵住事件循环。
import asyncio
import time
def read_file(filename):
time.sleep(2)
return f"读取完成：{filename}"
async def main():
result = await asyncio.to_thread(
read_file,
"data.txt",
)
print(result)
asyncio.run(main())
