10
并发与异步
threading / multiprocessing / asyncio
为什么你的程序在等网络、等文件的时候傻站着？因为它在"串行干活"。并发让它能同时干几件事。但 Python 有个著名的 GIL，选错工具白忙活。这章给你一张"什么时候用什么"的决策表。
论GIL：全局解释器锁
CPython（官方 Python 解释器）有一把全局锁：同一时刻，一个进程里只有一个线程在跑 Python 字节码。这意味着多线程并不能加速 CPU 密集型任务——多个线程抢同一把锁，等于排队。但 IO 密集型（等网络、等磁盘）不受影响，因为等的时候锁会释放，别的线程能用。
三种并发武器对照表
| 武器 | 适合 | 不适合 |
|---|
| 多线程 threading | IO 密集型（爬虫、读文件、调 API） | CPU 密集型（GIL 挡着）。 |
| 多进程 multiprocessing | CPU 密集型（图像计算、数值计算） | IO 密集型（进程创建贵）。 |
| 异步 asyncio | 超高并发 IO（成千上万网络请求） | CPU 密集；且必须用异步库。 |
concurrent.futures：最省心的并发
import time
from concurrent.futures import ThreadPoolExecutor
def fetch(url):
time.sleep(0.5) # 假装在等网络
return f"{url} 完成"
urls = ["a.com", "b.com", "c.com", "d.com"]
# 串行：要 2 秒
start = time.time()
for u in urls:
fetch(u)
print("串行：", time.time() - start)
# 线程池：4 个一起跑，0.5 秒搞定
start = time.time()
with ThreadPoolExecutor(max_workers=4) as pool:
results = list(pool.map(fetch, urls))
print("并发：", time.time() - start)
print(results)
async / await 异步
import asyncio
# async def 定义协程函数，调用它不会立刻执行，返回一个协程对象
async def fetch(name, delay):
print(f"{name} 开始等")
await asyncio.sleep(delay) # 异步等，不占线程
print(f"{name} 完成")
return name
async def main():
# 同时跑三个协程
results = await asyncio.gather(
fetch("A", 1),
fetch("B", 2),
fetch("C", 1.5),
)
return results
# Python 3.7+：asyncio.run() 启动事件循环
out = asyncio.run(main())
print(out) # ['A', 'B', 'C']
# 总耗时约 2 秒（最慢的那个），不是 4.5 秒
# 异步 HTTP 用 aiohttp（要 pip install aiohttp）
异步并发量控制：Semaphore
同时发 1000 个请求会把对方服务器打挂（也会把你自己封 IP）。用信号量限制"同时最多 10 个"：
async def fetch_limited(name, sem):
async with sem: # 同时只有 10 个能进来
await fetch(name, 1)
async def main():
sem = asyncio.Semaphore(10)
await asyncio.gather(*[fetch_limited(f"req{i}", sem)
for i in range(100)])
完整案例：aiohttp 异步爬虫（同步 vs 异步速度对比）
光说不练假把式。下面用 aiohttp 并发抓 10 个网页，对比同步 requests 和异步 aiohttp 的耗时差。这是异步最能打的场景——爬虫等网络的时间被压缩到几乎为零。
先装库：pip install aiohttp requests
import asyncio
import time
import requests
import aiohttp
URLS = [
"https://httpbin.org/delay/1", # 故意让服务器延迟 1 秒返回
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/1",
] * 2 # 一共 10 个请求
# ===== 同步版：requests，一个一个等 =====
def fetch_sync(url):
resp = requests.get(url, timeout=10)
return resp.status_code
start = time.time()
sync_results = [fetch_sync(u) for u in URLS]
sync_cost = time.time() - start
print(f"同步 requests：{sync_cost:.2f} 秒，结果 {len(sync_results)} 条")
# ===== 异步版：aiohttp + Semaphore 限流 =====
async def fetch_async(session, url, sem):
async with sem: # 同时最多 5 个请求，别把服务器打挂
async with session.get(url, timeout=10) as resp:
return resp.status
async def main_async():
sem = asyncio.Semaphore(5)
async with aiohttp.ClientSession() as session:
tasks = [fetch_async(session, u, sem) for u in URLS]
return await asyncio.gather(*tasks)
start = time.time()
async_results = asyncio.run(main_async())
async_cost = time.time() - start
print(f"异步 aiohttp：{async_cost:.2f} 秒，结果 {len(async_results)} 条")
print(f"提速 {sync_cost / async_cost:.1f} 倍")
运行输出（你机器上数字略有不同，但比例惊人）
# 同步 requests：10.23 秒，结果 10 条
# 异步 aiohttp：2.14 秒，结果 10 条
# 提速 4.8 倍
# 解析：同步版每个请求等 1 秒，10 个就是 10 秒；
# 异步版 5 个一批同时等，两批就完事，约 2 秒。
# 请求越多、延迟越高，异步优势越夸张——几百个请求能从 10 分钟压到 10 秒。
异步里别写阻塞代码
在 async def 里调用 requests.get()、time.sleep() 这种阻塞函数，会卡住整个事件循环，并发直接废掉。规则：异步代码里一切都要用异步版本——等网络用 aiohttp，等时间用 asyncio.sleep()，读文件用 aiofiles，连数据库用异步驱动（async SQLAlchemy）。偶尔必须调同步库？丢到 asyncio.to_thread() 里跑。
concurrent.futures 完整案例：map / submit / as_completed
ThreadPoolExecutor（线程池，IO 密集）和 ProcessPoolExecutor（进程池，CPU 密集）是 concurrent.futures 的两大门将。三种提交方式：map（最简单，按顺序返回）、submit（单个提交）+ as_completed（谁先完成先处理谁）。
import time
from concurrent.futures import (
ThreadPoolExecutor, ProcessPoolExecutor, as_completed
)
def work(n):
time.sleep(0.5 + n * 0.1) # 模拟每个任务耗时不同
return n * n
nums = [1, 2, 3, 4, 5]
# ① map：按输入顺序返回结果（哪怕 5 号先完成，也得等 1 号排到）
with ThreadPoolExecutor(max_workers=3) as pool:
results = list(pool.map(work, nums))
print(f"map 顺序结果：{results}")
# map 顺序结果：[1, 4, 9, 16, 25] 始终按 nums 的顺序
# ② submit + as_completed：谁先算完先用谁，不等慢的
with ThreadPoolExecutor(max_workers=3) as pool:
future_to_n = {pool.submit(work, n): n for n in nums}
for fut in as_completed(future_to_n):
n = future_to_n[fut]
print(f"任务 {n} 完成，结果 {fut.result()}")
# 任务 1 完成，结果 1
# 任务 2 完成，结果 4
# 任务 3 完成，结果 9
# 任务 4 完成，结果 16
# 任务 5 完成，结果 25 （顺序取决于谁先跑完，这里恰好从小到大）
# ③ ProcessPoolExecutor：CPU 密集型用这个（绕开 GIL）
def cpu_heavy(n):
s = 0
for i in range(1_0000_000):
s += i * n
return s
with ProcessPoolExecutor(max_workers=4) as pool:
total = sum(pool.map(cpu_heavy, [1, 2, 3, 4]))
print(f"CPU 密集多进程结果：{total}")
论map 和 as_completed 怎么选
map：结果顺序和输入一致，代码最干净，适合"我只要全部结果，不关心谁先完成"。submit + as_completed：谁先干完就先处理谁（比如先渲染先下载的图片），适合流式展示、超时取消。CPU 密集型把 ThreadPoolExecutor 换成 ProcessPoolExecutor 即可，接口一模一样——这就是 concurrent.futures 的优雅之处。
本章面试题 · 并发与异步
1.（概念题）什么是 GIL？它对多线程和多进程分别有什么影响？
查看答案
答案：GIL（全局解释器锁）是 CPython 保证同一时刻只有一个线程执行 Python 字节码的机制。多线程在 CPU 密集任务上无法真正并行（互相抢锁），但 IO 密集任务上有效（等 IO 时释放锁）；多进程每个进程有独立 GIL，能真正利用多核，适合 CPU 密集，但进程创建开销大、内存不共享。
2.（概念题）asyncio 和多线程有什么区别？为什么异步爬虫比多线程还快？
查看答案
答案：多线程靠操作系统切换线程，有内核态开销；asyncio 在单个线程里由事件循环调度协程，切换是用户态的，极轻量。爬虫等网络时两者都在"等"，但 asyncio 一个线程就能挂起几千个协程，多线程开几千个线程上下文切换成本很高。且 asyncio 没有 GIL 抢锁问题。
3.（代码题）下面这段异步代码有什么问题？
async def main():
results = [await fetch(u) for u in URLs]
asyncio.run(main())
查看答案
答案：这是串行异步——列表推导里 await 一个接一个，等于没用上并发。正确做法：先创建所有协程/任务，再 gather：tasks = [fetch(u) for u in URLs]; await asyncio.gather(*tasks)。记住：await 是"等这一个"，gather 才是"一起跑"。
4.（概念题）ThreadPoolExecutor 的 map 和 as_completed 有什么区别？
查看答案
答案：map 返回结果顺序和输入一致，哪怕第 5 个任务先完成，也要等前面的排到；submit + as_completed 按完成先后迭代，谁先返回先处理谁。要保顺序用 map，要流式处理用 as_completed。
5.（思考题）为什么在 async def 里直接调 requests.get() 会"毁掉"异步性能？怎么补救？
查看答案
答案：requests.get() 是阻塞调用，一旦它在等网络，整个事件循环都被它卡住，其他协程全部停摆，并发形同虚设。补救：① 换成异步库 aiohttp；② 实在要用同步库，丢进 await asyncio.to_thread(requests.get, url)，让它在线程里跑，不阻塞事件循环。