Python 并发怎么选？多线程 / 多进程 / asyncio 真实性能对比（附实测数据 + 选型决策图）

Python 并发怎么选？多线程 / 多进程 / asyncio 真实性能对比（附实测数据 + 选型决策图）

📑 目录

• 摘要：一个反直觉的实测结果
• 一、三种并发方案 30 秒扫盲
• 二、测试设计：怎么测才可信
• 三、IO 密集场景实测：asyncio 完胜
• 四、CPU 密集场景实测：多线程居然不加速
• 五、混合场景实测：谁才是万金油
• 六、内存开销对比：多进程的隐藏成本
• 七、原理深挖：为什么数据长这样
• 八、可复现代码（完整脚本）
• 九、选型决策图 + 避坑清单
• 总结

摘要：一个反直觉的实测结果

很多 Python 开发者听过一句话：“IO 密集用多线程，CPU 密集用多进程，高并发用 asyncio。” 但真到选型时，三个问题答不上来：

网上教程大多只甩结论，不甩数据。笔者写了一个完整 benchmark，三大场景 × 四种方案 × 三个量级，在 20 核 Windows + Python 3.13 上跑出真实数据。先看一个最反直觉的结果：

CPU 密集任务（统计质数），多线程耗时 10.53s，串行耗时 10.54s——多线程几乎没有加速。 而同样任务用多进程，只要 1.72s，快了 6 倍。

这不是写法问题，是 GIL（全局解释器锁） 的铁证。下面用数据把三种方案彻底讲透。

一、三种并发方案 30 秒扫盲

一句话区分：

• 多线程：多个工人挤在一个车间（GIL），同一时刻只有一个能干活，但等人（IO）时可以让别人干
• 多进程：多个独立车间，真正同时干活，但车间建造成本高（内存大）
• asyncio：一个工人快速切换多个任务，等人（IO）时立刻切下一个，切换成本最低

二、测试设计：怎么测才可信

2.1 测试环境

2.2 三大场景定义

2.3 四种方案 × 三个量级

• 方案：串行 / 多线程 / 多进程 / asyncio
• 量级：N=50 / 200 / 500 个任务
• 指标：总耗时（秒）、峰值内存增量（MB）

量级选 50/200/500 是因为：50 看趋势，200 是日常常见量，500 验证高并发下的稳定性。CPU 场景在 500 时太慢，统一截到 200。

三、IO 密集场景实测：asyncio 完胜

3.1 实测数据

任务：N 个 HTTP 请求，每个服务端处理 0.2s。

3.2 数据解读

三个结论，数据说话：

3.3 趋势：量级越大，asyncio 与多线程差距缩小

结论：小量级 IO 任务，asyncio 优势明显；大量级时两者接近，但 asyncio 内存更省、扩展性更好。

四、CPU 密集场景实测：多线程居然不加速

4.1 实测数据

任务：N 次统计 [2, 80000] 内质数。

4.2 数据解读

这是整篇文章最关键的一组数据：

4.3 为什么多线程没加速？GIL 一句话讲清

GIL（Global Interpreter Lock）：CPython 解释器在同一时刻只允许一个线程执行 Python 字节码。多线程在 CPU 密集任务下，看似多线程，实际还是轮流跑，还要额外付线程切换开销，所以不仅不加速，有时还更慢。

GIL 只在 纯 Python 计算时 持有。如果 CPU 任务是 C 扩展（如 NumPy 矩阵运算），GIL 会被释放，多线程可以加速——但纯 Python 循环计算，GIL 一锁死。

五、混合场景实测：谁才是万金油

5.1 实测数据

任务：每个任务 = 算小范围质数 + 请求一次 HTTP。

5.2 数据解读

混合场景最贴近真实业务（取数 + 计算 + 落库）：

结论：真实业务（IO + CPU 混合）首选 asyncio + executor，次选多线程，多进程只在大计算量时才考虑。

六、内存开销对比：多进程的隐藏成本

6.1 峰值内存增量（N=200，单位 MB）

6.2 重要说明：多进程的真实内存远高于表格

上表测的是主进程的内存增量。多进程方案下，每个子进程都是独立的 Python 解释器，真实总内存 = 主进程 + N × 子进程内存。

实测：一个空载 Python 子进程约 15~20MB，8 个子进程额外占用 120~160MB，是 asyncio 的 100 倍以上。

结论：多进程用性能换内存。CPU 密集任务必须用，但要注意服务器内存上限，别开太多进程把内存撑爆。

七、原理深挖：为什么数据长这样

7.1 三张图看懂 GIL、进程、协程

多线程（受 GIL 限制）：

时间轴 →
线程1: [计算██] [等IO ] [计算██] [等IO ]
线程2: [等IO ] [计算██] [等IO ]
↑ GIL 在 IO 时释放, 所以 IO 能并发
↑ 但计算时 GIL 锁住, 只有一个线程能算

多进程（无 GIL，真并行）：

时间轴 →
进程1: [计算████████]
进程2: [计算████████] ← 真正同时计算, 各自独立 GIL
进程3: [计算████████]
进程4: [计算████████]

asyncio（单线程事件循环）：

时间轴 →
主线程: [算] [等IO→切] [算] [等IO→切] [算] [等IO→切]
↑ IO 等待时不阻塞, 立刻切下一个任务
↑ 但计算时独占, 无并发

7.2 为什么 IO 场景三者都加速？

IO 操作（网络请求、磁盘读写）会释放 GIL。所以多线程在 IO 等待时，其他线程能拿到 GIL 干活，实现并发。asyncio 更彻底，连线程切换都省了，直接在单线程内切换协程，效率更高。

7.3 为什么 CPU 场景只有多进程加速？

纯 Python 计算不释放 GIL。多线程抢 GIL，同一时刻只有一个线程在算，加切换开销后≈串行。多进程每个进程有独立 GIL，真正并行计算。

7.4 何时多线程也能加速 CPU？

当 CPU 计算是 C 扩展（如 NumPy、Pandas 的底层运算）时，GIL 会被释放，多线程可以加速。所以 NumPy 矩阵运算用多线程是有效的，但纯 Python for 循环计算用多线程无效。

八、可复现代码（完整脚本）

下面是本文所有数据的来源脚本，复制即可运行，已处理 Windows 多进程兼容性：

"""
Python 并发方案性能对比 benchmark
四种方案: 串行 / 多线程 / 多进程 / asyncio
三大场景: IO密集 / CPU密集 / 混合
"""
import time, os, sys, asyncio, threading, multiprocessing as mp
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import psutil, aiohttp

sys.stdout.reconfigure(encoding="utf-8")
PROC = psutil.Process(os.getpid())
def mem_mb(): return PROC.memory_info().rss / 1024 / 1024

PORT = 18923
URL = f"http://127.0.0.1:{PORT}/"

class SlowHandler(BaseHTTPRequestHandler):
def do_GET(self):
time.sleep(0.2) # 模拟服务端处理
self.send_response(200); self.end_headers(); self.wfile.write(b"ok")
def log_message(self, *a): pass

# 顶层函数 (multiprocessing spawn 需要)
def count_primes(limit=80000):
c = 0
for i in range(2, limit):
if all(i % j for j in range(2, int(i**0.5)+1)): c += 1
return c

def _io_one(_):
import urllib.request; urllib.request.urlopen(URL, timeout=5).read()
def _cpu_one(_): return count_primes(80000)
def _mix_one(_):
count_primes(20000)
import urllib.request; urllib.request.urlopen(URL, timeout=5).read()

# —- IO 场景 —-
def io_serial(n):
import urllib.request
for _ in range(n): urllib.request.urlopen(URL, timeout=5).read()
def io_thread(n, w=20):
with ThreadPoolExecutor(w) as ex: list(ex.map(_io_one, range(n)))
def io_process(n, w=8):
with ProcessPoolExecutor(w) as ex: list(ex.map(_io_one, range(n)))
async def io_async(n, w=20):
sem = asyncio.Semaphore(w)
async with aiohttp.ClientSession() as s:
async def one():
async with sem, s.get(URL) as r: await r.read()
await asyncio.gather(*[one() for _ in range(n)])

# —- CPU 场景 —-
def cpu_serial(n):
for _ in range(n): count_primes(80000)
def cpu_thread(n, w=8):
with ThreadPoolExecutor(w) as ex: list(ex.map(_cpu_one, range(n)))
def cpu_process(n, w=8):
with ProcessPoolExecutor(w) as ex: list(ex.map(_cpu_one, range(n)))
async def cpu_async(n, w=8):
loop = asyncio.get_event_loop()
with ThreadPoolExecutor(w) as ex:
await asyncio.gather(*[loop.run_in_executor(ex, count_primes, 80000) for _ in range(n)])

# —- 运行器 —-
def bench(name, fn, *args):
m0, t0 = mem_mb(), time.perf_counter()
try:
asyncio.run(fn(*args)) if asyncio.iscoroutinefunction(fn) else fn(*args)
err = None
except Exception as e: err = str(e)[:50]
t1, m1 = time.perf_counter(), mem_mb()
return name, round(t1–t0,3) if not err else None, round(m1–m0,2) if not err else None, err

def main():
srv = ThreadingHTTPServer(("127.0.0.1", PORT), SlowHandler)
srv.daemon_threads = True
threading.Thread(target=srv.serve_forever, daemon=True).start()
time.sleep(0.5)
print(f"CPU: {os.cpu_count()} cores | Python {sys.version.split()[0]}\\n")
for N in [50, 200, 500]:
print(f"=== N={N} ===")
print("[IO]", [bench(*x) for x in [("serial",io_serial,N),("thread",io_thread,N),("process",io_process,N),("asyncio",io_async,N)]])
cn = min(N, 200)
print("[CPU]", [bench(*x) for x in [("serial",cpu_serial,cn),("thread",cpu_thread,cn),("process",cpu_process,cn),("asyncio",cpu_async,cn)]])
srv.shutdown()

if __name__ == "__main__":
mp.freeze_support()
main()

运行方式：python bench_concurrency.py，需要 pip install psutil aiohttp。

九、选型决策图 + 避坑清单

9.1 选型决策图

你的任务主要是什么?
│
├─ IO 密集 (网络请求/数据库/文件读写)
│ ├─ 并发量 < 1000 → 多线程 (threading) [简单, 生态好]
│ └─ 并发量 ≥ 1000 → asyncio [最省资源, 最快]
│
├─ CPU 密集 (数值计算/加密/图像处理)
│ ├─ 用 NumPy/Pandas → 多线程也可 (C扩展释放GIL)
│ └─ 纯 Python 计算 → 多进程 (multiprocessing) [唯一解]
│
└─ 混合 (IO + CPU)
└─ asyncio + run_in_executor [IO用协程, CPU丢线程池]

9.2 五大避坑清单

9.3 一张表总结

总结

本文用 36 组实测数据（3 场景 × 4 方案 × 3 量级）回答了 Python 并发选型的三个核心问题：

最终选型一句话：IO 多用 asyncio，CPU 多用多进程，混合用 asyncio + executor，简单脚本用多线程。

本文所有数据均可复现，完整脚本见第八节。测试环境：20 核 / Python 3.13.12 / Windows 11。不同机器数值会变，但结论（GIL 限制、asyncio IO 优势、多进程内存代价）不变。