Python►并发编程

异步编程 asyncio 入门

asyncio 是 Python 3.4+ 引入、3.7+ 大幅简化的异步编程框架。它让单线程处理上万个并发 IO 任务成为可能——网络爬虫、Web 服务、聊天服务器、实时数据流都是它的主场。这一篇讲清楚：什么是协程、async/await 怎么写、asyncio.run / gather / create_task 的区别、以及为什么 asyncio 比多线程更适合大并发。

一、异步的核心思想

同步：一次只做一件事，卡住就干等。

异步：一件事等 IO 时主动让出控制权，去做别的事，IO 好了再回来。

想象你在餐厅点了 3 道菜：

  • 同步：点第 1 道，坐着等做好；再点第 2 道，坐着等……做完 3 道要 30 分钟；
  • 异步：一次性下 3 道单，等哪道好了就上哪道。10 分钟就吃完。

asyncio 就是让你的程序像”下多道单”一样。

二、协程：async def

1
2
3
4
5
6
7
8
import asyncio

async def hello():
    print("开始")
    await asyncio.sleep(1)      # 让出 1 秒
    print("结束")

asyncio.run(hello())
  • async def 定义协程函数；
  • 调用它得到协程对象（不会立即执行）；
  • await 等待另一个协程完成（期间让出 CPU 给别的协程）；
  • asyncio.run(coro) 启动事件循环，运行协程。

注意：不能在 async def 之外用 await（会 SyntaxError）。

三、并发运行：asyncio.gather

想让多个协程并发跑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio
import time


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


async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )
    print(f"耗时 {time.perf_counter() - start:.2f}s，结果 {results}")


asyncio.run(main())

输出：

1
2
3
4
5
6
7
A 开始
B 开始
C 开始
B 结束
A 结束
C 结束
耗时 3.00s，结果 ['A', 'B', 'C']

3 个任务并发跑，总耗时等于最慢的那个（不是三个相加）。

四、create_task：立即调度

1
2
3
4
5
async def main():
    task1 = asyncio.create_task(task("A", 2))     # 立即调度
    task2 = asyncio.create_task(task("B", 1))
    result1 = await task1
    result2 = await task2

create_task 让协程立即开始，await 只是等它结束。

gather vs create_task：

  • 想统一等一批任务全部完成 → gather；
  • 想任务立即跑，之后再取结果 → create_task。

五、asyncio.sleep vs time.sleep

永远别在 async 函数里用 time.sleep——它会阻塞整个事件循环：

1
2
3
4
5
async def bad():
    time.sleep(1)         # ❌ 整个事件循环卡 1 秒

async def good():
    await asyncio.sleep(1)   # ✅ 让出控制

六、await 只能等”可等待对象”

  • 协程（async def 定义的函数调用返回）；
  • Task（create_task 返回）；
  • Future（底层原语）；
  • 实现了 __await__ 的自定义对象。

普通函数调用不能 await：

1
2
3
4
async def main():
    time.sleep(1)           # ❌ 阻塞，不 await
    await time.sleep(1)     # ❌ time.sleep 返回 None，不能 await
    await asyncio.sleep(1)  # ✅

七、异步 HTTP：httpx / aiohttp

爬 100 个 URL 的对比：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 顺序 requests
import requests
for url in urls:
    r = requests.get(url)
# 100 秒

# 异步 httpx
import httpx
import asyncio

async def fetch(client, url):
    r = await client.get(url)
    return r.status_code

async def main():
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[fetch(client, u) for u in urls])
    return results

asyncio.run(main())
# 1-2 秒

100 个并发几乎没成本——单线程！

八、超时

1
2
3
4
5
6
7
8
9
try:
    await asyncio.wait_for(some_coro(), timeout=5)
except asyncio.TimeoutError:
    print("超时")


# Python 3.11+ 更优雅
async with asyncio.timeout(5):
    await some_coro()

九、错误处理

协程里的异常会传播到 await 处：

1
2
3
4
5
6
7
8
async def bad():
    raise ValueError("boom")

async def main():
    try:
        await bad()
    except ValueError as e:
        print(f"抓到：{e}")

gather 默认在任何一个协程抛异常时立即抛出，其他协程被取消：

1
2
3
4
5
6
7
8
try:
    await asyncio.gather(bad(), good())
except ValueError:
    ...

# 不想让一个失败拖累全部：
results = await asyncio.gather(bad(), good(), return_exceptions=True)
# results 里失败的项是异常对象

十、TaskGroup（Python 3.11+）

新语法，比 gather 更结构化：

1
2
3
4
5
6
async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task("A", 2))
        t2 = tg.create_task(task("B", 1))
    # 出块时会等所有 task 完成
    # 一个抛异常会取消其他，最后一起 ExceptionGroup

十一、Semaphore：限并发

不想同时 10000 个请求打爆对方：

1
2
3
4
5
6
7
8
9
sem = asyncio.Semaphore(20)      # 最多 20 个并发

async def fetch(url):
    async with sem:
        # 同时最多 20 个进这里
        return await client.get(url)


await asyncio.gather(*[fetch(u) for u in urls])

十二、锁 / 事件 / 队列

跟 threading 类似，asyncio 也有对应的同步原语：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
lock = asyncio.Lock()
event = asyncio.Event()
queue = asyncio.Queue()

async def worker(q):
    while True:
        item = await q.get()
        if item is None:
            break
        process(item)
        q.task_done()

# 生产者-消费者
async def main():
    q = asyncio.Queue()
    workers = [asyncio.create_task(worker(q)) for _ in range(5)]
    for item in items:
        await q.put(item)
    for _ in workers:
        await q.put(None)
    await asyncio.gather(*workers)

十三、在同步代码里跑异步

1
2
3
4
# ✅ 一次性入口
asyncio.run(main())

# ❌ 在已有 event loop 里再 asyncio.run 会报错

Jupyter Notebook 里已经有 loop，用 await 直接跑或 nest_asyncio 补丁。

十四、把同步代码”变异步”

有些库只有同步 API，不能 await。放到线程池里跑：

1
2
3
4
5
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, blocking_func, arg1, arg2)

# 或者 Python 3.9+：
result = await asyncio.to_thread(blocking_func, arg1, arg2)

这样阻塞的 requests 也能在异步程序里用（但慢得多，能用 httpx 就用 httpx）。

十五、asyncio vs threading vs multiprocessing

场景 推荐
大量 IO 并发（爬虫、Web、DB） asyncio
现有同步库为主的项目 threading
CPU 密集计算 multiprocessing
简单的”下 20 个请求” threading（更简单）
长期运行的服务器（10000+ 并发） asyncio

十六、常见异步库

  • httpx：HTTP，同步 + 异步；
  • aiohttp：HTTP + WebSocket；
  • aiofiles：文件 IO；
  • asyncpg、aiomysql、motor：数据库；
  • FastAPI：异步 Web 框架；
  • redis-py（4.x+）：Redis 支持 async。

十七、一个完整例子：并发爬虫

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# crawler.py
"""异步并发爬取多个 URL，展示 asyncio 的力量。"""

import asyncio
import httpx
import time


async def fetch(client, url, sem):
    async with sem:
        try:
            r = await client.get(url, timeout=10)
            return url, r.status_code, len(r.text)
        except Exception as e:
            return url, None, str(e)


async def main(urls, concurrency=20):
    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, u, sem) for u in urls]
        return await asyncio.gather(*tasks)


if __name__ == "__main__":
    urls = [f"https://httpbin.org/status/{code}" for code in range(200, 300)]
    start = time.perf_counter()
    results = asyncio.run(main(urls))
    print(f"共 {len(results)} 个，耗时 {time.perf_counter() - start:.2f}s")
    ok = sum(1 for u, s, _ in results if s == 200)
    print(f"200 状态：{ok}")

100 个请求几秒钟就能跑完。

十八、常见陷阱

  1. 在 async 函数里用 time.sleep：卡死事件循环。用 asyncio.sleep。
  2. await 一个普通函数：报错。只能 await 协程/Task/Future。
  3. 协程创建了不 await：RuntimeWarning: coroutine never awaited。
  4. 在同步函数里用 async：只能通过 asyncio.run 入口。
  5. 一个协程被 await 多次：只能 await 一次；多次 await 用 create_task 包装。
  6. gather 里一个失败全挂：加 return_exceptions=True。
  7. CPU 密集协程：await 不会自动让出，一直占用事件循环。用 run_in_executor。

十九、小结与延伸阅读

  • async def 定义协程，await 等待；
  • asyncio.run(main()) 是入口；
  • gather 并发跑一批，create_task 立即调度；
  • 用异步 IO 库（httpx、aiohttp、asyncpg）；
  • 阻塞函数用 asyncio.to_thread；
  • 限并发用 asyncio.Semaphore；
  • 超时用 asyncio.timeout 或 wait_for；
  • IO 密集用 asyncio；CPU 密集用 multiprocessing。

延伸阅读：

下一篇也是最后一篇 类型注解 type hints 实战 我们讲怎么让 Python 代码”看起来是静态类型语言”。