Python 并发编程入门：多线程、多进程、异步怎么选

先搞懂 GIL：Python 并发的总开关

GIL（全局解释器锁）是 CPython 的一个机制：同一时刻，一个进程里只允许一个线程执行 Python 字节码。

带来的直接后果：

• CPU 密集型任务（计算、图像处理）：多线程在 Python 里没用，甚至更慢（多了切换开销）
• IO 密集型任务（网络请求、读写文件、数据库）：多线程有效！因为线程在等待 IO 时会释放 GIL，别的线程趁机干活

判断你的任务类型，方案就选对了一半：

任务类型	例子	正确方案
IO 密集	爬虫、调 API、读写数据库	多线程 或 asyncio
CPU 密集	视频转码、科学计算、数据加密	多进程

方案一：多线程（IO 密集场景的简单解）

串行 vs 多线程下载 10 个网页：

import time
from concurrent.futures import ThreadPoolExecutor

import requests

URLS = ["https://httpbin.org/delay/1"] * 10  # 每个请求固定延迟1秒

def fetch(url: str) -> int:
    resp = requests.get(url, timeout=10)
    return len(resp.content)

# 串行：约 10 秒
start = time.time()
for u in URLS:
    fetch(u)
print(f"串行耗时：{time.time() - start:.1f}s")

# 10 线程：约 1 秒
start = time.time()
with ThreadPoolExecutor(max_workers=10) as pool:
    list(pool.map(fetch, URLS))
print(f"多线程耗时：{time.time() - start:.1f}s")

10 倍的差距，代码只改了三行。ThreadPoolExecutor.map 是日常最够用的写法：把任务列表扔进去，拿回结果列表，线程池帮你搞定创建和回收。

线程安全提醒：多个线程同时改同一个列表/字典会出诡异 bug。要么用 queue.Queue（线程安全队列），要么给共享数据加锁：

import threading

lock = threading.Lock()
counter = 0

def add():
    global counter
    for _ in range(100000):
        with lock:          # 加锁，同一时刻只有一个线程能改
            counter += 1

方案二：多进程（CPU 密集的唯一解）

import time
from concurrent.futures import ProcessPoolExecutor

def heavy_compute(n: int) -> int:
    """模拟重计算：算第 n 个斐波那契数（故意用低效递归）"""
    if n < 2:
        return n
    return heavy_compute(n - 1) + heavy_compute(n - 2)

TASKS = [32] * 8

# 串行
start = time.time()
for t in TASKS:
    heavy_compute(t)
print(f"串行：{time.time() - start:.1f}s")

# 8 进程（吃满多核 CPU）
start = time.time()
with ProcessPoolExecutor(max_workers=8) as pool:
    list(pool.map(heavy_compute, TASKS))
print(f"多进程：{time.time() - start:.1f}s")

接口和线程池几乎一样（ProcessPoolExecutor），但底层是开多个 Python 进程，每个进程有独立的 GIL，所以能真正吃满多核。

代价：进程开销比线程大得多（内存、启动慢），进程间不能直接共享变量，要通信用 multiprocessing.Queue 或共享内存。进程数别乱开，一般等于 CPU 核数。

方案三：asyncio（高并发 IO 的终极形态）

多线程的问题是线程多了开销大、切换频繁。asyncio 用单线程 + 协程实现并发：一个线程里，哪个协程在等 IO 就切走，不切操作系统线程，切换成本几乎为零。

单机轻松维持上万并发连接——这是线程池做不到的。

import asyncio
import time

import aiohttp

URLS = ["https://httpbin.org/delay/1"] * 20

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as resp:
        data = await resp.read()
        return len(data)

async def main():
    async with aiohttp.ClientSession() as session:
        # 并发发起所有请求
        tasks = [fetch(session, u) for u in URLS]
        results = await asyncio.gather(*tasks)
        print(f"完成 {len(results)} 个请求")

start = time.time()
asyncio.run(main())   # 20 个请求约 1 秒完成，且只用了 1 个线程
print(f"异步耗时：{time.time() - start:.1f}s")

关键语法：

• async def 定义协程函数
• await 表示「这里要等 IO，先让别的协程跑」
• asyncio.gather 并发执行一批协程
• 网络库要用异步版：aiohttp 替代 requests

控制并发量（别把目标服务器打爆）：

sem = asyncio.Semaphore(5)  # 最多同时5个

async def fetch_limited(session, url):
    async with sem:   # 拿到令牌才能执行
        return await fetch(session, url)

三方案对比实测（20 个延迟 1 秒的请求）

方案	耗时	线程/进程数	适合场景
串行	20s	1	任务少且快
多线程(10)	2s	10	几十个并发，代码简单优先
asyncio	1s	1（协程并发）	成百上千并发，追求极致
多进程(8)	≈20s	8	这是 IO 任务，多进程帮不上忙

怎么选：决策流程

任务主要是等待（网络/磁盘/数据库）？
├── 是 → 并发量多大？
│   ├── 几十上百 → ThreadPoolExecutor（简单）
│   └── 成千上万 → asyncio + aiohttp（高性能）
└── 否（CPU 计算为主）→ ProcessPoolExecutor
    （或更优：用 numpy 这类底层 C 实现的库，天然绕开 GIL）

常见坑

1. 在协程里调了同步阻塞函数

async def bad():
    time.sleep(1)        # 错误！整个事件循环卡住1秒
    await asyncio.sleep(1)  # 正确：让出控制权

同理：协程里用 requests（同步库）会卡死事件循环，必须用 aiohttp。

2. 共享数据没加保护

asyncio 单线程看似安全，但 await 点之间数据可能被其他协程改掉。跨 await 的复合操作仍需 asyncio.Lock。

3. 盲目开几百个线程

线程不是越多越好，IO 任务开到几十个收益就见顶，再多只剩切换开销和内存浪费。

写在最后

并发编程的复杂度不在 API 而在思维方式：从「一步一步来」切换到「同时推进很多事」。建议的入门路径：先把 ThreadPoolExecutor.map 用到熟（能解决 80% 的需求），需要高并发时再学 asyncio，CPU 瓶颈出现时自然懂得上多进程。