把"线程"和"进程"装进池子里，让Python并发像写同步代码一样简单
0. 为什么选concurrent.futures？
| 方案 | 易用性 | 自动复用 | 返回值 | 异常捕获 |
|---|---|---|---|---|
| threading | 低(手动join) | ❌ | 手动 | 易漏 |
| multiprocessing | 低(Queue) | ❌ | 手动 | 易漏 |
| concurrent.futures | ⭐⭐⭐ | ✅ | Future对象 | 统一 |
1. 两大池子概览
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
| 池子 | 适用场景 | 并行度 | 开销 | GIL |
|---|---|---|---|---|
| ThreadPoolExecutor | IO密集(网络/磁盘) | ≈CPU核数 | 低(共享内存) | 受GIL |
| ProcessPoolExecutor | CPU密集(计算) | =CPU核数 | 高(独立解释器) | 绕过GIL |
2. 10行代码跑通线程池
import requests, time
from concurrent.futures import ThreadPoolExecutor, as_completedurls = [f"https://httpbin.org/delay/{i}" for i in range(1, 6)]def fetch(url):return requests.get(url, timeout=10).status_codestart = time.time()
with ThreadPoolExecutor(max_workers=5) as pool:futures = [pool.submit(fetch, u) for u in urls]for f in as_completed(futures): # 完成顺序返回print(f.result(), time.time()-start)
vs串行：5次×1s → 并发1.2s，提速4倍|tddh
3. 核心API一图流
submit(fn, *args, **kwargs) → Future└→ future.result(timeout) / exception() / cancel()map(func, *iterables, chunksize=1) → 迭代器(按输入顺序)as_completed(futures) → 迭代器(完成顺序)
Future对象常用方法
| 方法 | 说明 |
|---|---|
result(timeout=None) |
阻塞获取返回值 |
exception(timeout=None) |
获取异常对象 |
cancel() |
取消未运行任务 |
add_done_callback(fn) |
完成后回调 |
4. 进程池演示：CPU密集
import math, time
from concurrent.futures import ProcessPoolExecutordef is_prime(n):if n < 2: return Falsefor i in range(2, int(math.sqrt(n)) + 1):if n % i == 0: return Falsereturn Truenumbers = list(range(1, 100000))start = time.time()
with ProcessPoolExecutor(max_workers=8) as pool:results = pool.map(is_prime, numbers, chunksize=1000)list(results) # 消耗迭代器
print("ProcessPool cost:", time.time() - start)
8核MacBook：串行12s → 进程池2.1s，5.7×加速
5. 异常处理：永不遗漏
def div(a, b):return a / bwith ThreadPoolExecutor() as pool:futures = [pool.submit(div, 10, i) for i in [1, 0, 2]]for f in futures:try:print(f.result())except ZeroDivisionError as e:print("捕获异常:", e)
6. 性能调优 checklist
| 参数 | 建议值 | 说明 |
|---|---|---|
| max_workers | CPU核数×2（IO）/核数（CPU） | 默认=min(32, os.cpu_count()+4) |
| chunksize | 1000-10000（ProcessPool.map） | 减少IPC次数 |
| thread_name_prefix | "PoolThread" | 方便调试 |
7. 与asyncio协作
import asyncio, aiohttp
from concurrent.futures import ThreadPoolExecutordef sync_fetch(url):return requests.get(url).textasync def main():loop = asyncio.get_event_loop()with ThreadPoolExecutor() as pool:tasks = [loop.run_in_executor(pool, sync_fetch, u) for u in urls]return await asyncio.gather(*tasks)asyncio.run(main())
把阻塞函数扔进线程池，不改造源码也能享受异步！
8. 实战：多线程下载+多进程压缩
import requests, PIL.Image as Image, os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutordef download(url):fname = url.split("/")[-1]with requests.get(url, stream=True) as r:with open(fname, "wb") as f:for chunk in r.iter_content(1024):f.write(chunk)return fnamedef thumbnail(fname, size=(128, 128)):img = Image.open(fname)img.thumbnail(size)out = "t_" + fnameimg.save(out)return out# 1. 线程池：IO密集下载
with ThreadPoolExecutor(max_workers=10) as tpool:files = list(tpool.map(download, url_list))# 2. 进程池：CPU密集压缩
with ProcessPoolExecutor() as ppool:ppool.map(thumbnail, files)
8核实测：100张4MB图片 → 下载30s+压缩15s，并行后总耗时35s
9. 常见坑与解决
| 坑 | 现象 | 解决 |
|---|---|---|
| 进程池pickle失败 | lambda/局部函数 | 定义在模块顶层 |
| 线程池GIL阻塞 | CPU任务无加速 | 换ProcessPool |
| 忘记消耗迭代器 | map返回生成器 | list()或for循环 |
| 取消已运行任务 | cancel()返回False | 判断future.running() |
10. 总结：一张思维导图
并发任务
├─ IO密集 → ThreadPoolExecutor
├─ CPU密集 → ProcessPoolExecutor
├─ 异步协程 → asyncio.run_in_executor
└─ 性能调优 → chunksize+max_workers
记住：先池化，再异步，最后量化验证——让并发变成一件优雅的事！