Python 并发模型对比：从 threading 到 asyncio 到多进程的性能边界

一、并发选型的困惑：为什么"用 asyncio 就对了"是错误答案

Python 并发编程有三个主流模型：threading（多线程）、asyncio（协程）和 multiprocessing（多进程）。社区中常见的一种观点是"I/O 密集用 asyncio，CPU 密集用多进程"。但实际场景远比这个二分法复杂：一个 Web 服务同时存在 I/O 等待和 CPU 计算，如何选择？asyncio 的回调地狱和调试困难如何解决？多进程的进程间通信开销如何控制？

更根本的问题是，Python 的 GIL（全局解释器锁）使得多线程无法真正并行执行 Python 字节码。这意味着 threading 仅适用于 I/O 密集场景（线程在 I/O 等待时释放 GIL），对 CPU 密集场景无能为力。

二、三种并发模型的底层机制与适用场景

flowchart TD
    subgraph "Threading 多线程"
        A1[Thread 1] --> B1[GIL 获取]
        A2[Thread 2] --> B2[GIL 等待]
        A3[Thread 3] --> B3[GIL 等待]
        B1 --> C1[执行字节码]
        C1 --> D1[I/O 等待 → 释放 GIL]
        D1 --> B2
    end

    subgraph "Asyncio 协程"
        E1[Task 1] --> F1[事件循环]
        E2[Task 2] --> F1
        E3[Task 3] --> F1
        F1 --> G1[await → 挂起当前协程]
        G1 --> H1[切换到就绪协程]
    end

    subgraph "Multiprocessing 多进程"
        I1[Process 1] --> J1[独立 GIL]
        I2[Process 2] --> J2[独立 GIL]
        I3[Process 3] --> J3[独立 GIL]
        J1 --> K1[真正并行]
        J2 --> K2[真正并行]
        J3 --> K3[真正并行]
    end

核心区别：threading 和 asyncio 都是并发（concurrency）而非并行（parallelism），同一时刻只有一个线程/协程在执行 Python 代码。multiprocessing 是真正的并行，每个进程有独立的 GIL。

三、工程实现与性能对比

3.1 Threading：I/O 并发与线程安全

import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from queue import Queue

class ThreadSafeCounter:
    """线程安全计数器，避免竞态条件"""
    def __init__(self):
        self._value = 0
        self._lock = threading.Lock()

    def increment(self):
        with self._lock:
            self._value += 1

    @property
    def value(self):
        with self._lock:
            return self._value


def fetch_urls_threading(urls, max_workers=10):
    """多线程并发请求"""
    results = {}
    counter = ThreadSafeCounter()

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(requests.get, url): url
            for url in urls
        }
        for future in as_completed(futures):
            url = futures[future]
            try:
                response = future.result(timeout=30)
                results[url] = response.status_code
                counter.increment()
            except Exception as e:
                results[url] = str(e)

    return results

3.2 Asyncio：高并发 I/O 的协程方案

import asyncio
import aiohttp

async def fetch_url(session, url, semaphore):
    """带信号量控制的异步请求"""
    async with semaphore:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(
                    total=30)) as response:
                return url, response.status
        except Exception as e:
            return url, str(e)

async def fetch_urls_async(urls, max_concurrent=100):
    """异步并发请求，支持更高并发度"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, url, semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    return dict(results)

# asyncio 的优势：单线程可处理数千并发连接
# threading 的上限通常在数百个线程（线程栈开销约 8MB/线程）

3.3 Multiprocessing：CPU 并行的进程池

from multiprocessing import Pool, shared_memory
import numpy as np

def cpu_intensive_task(args):
    """CPU 密集型任务：矩阵运算"""
    data, indices = args
    # 每个子进程有独立 GIL，可以真正并行
    result = np.linalg.svd(data)
    return result[1][:10]  # 返回前 10 个奇异值

def parallel_matrix_computation(matrices, num_processes=None):
    """多进程并行矩阵计算"""
    with Pool(processes=num_processes) as pool:
        results = pool.map(
            cpu_intensive_task,
            [(m, i) for i, m in enumerate(matrices)]
        )
    return results

# 进程间共享数据：使用共享内存避免序列化开销
def shared_memory_computation(data_shape):
    """基于共享内存的零拷贝并行计算"""
    shm = shared_memory.SharedMemory(create=True, size=1024*1024*100)

    # 子进程通过共享内存访问数据，无需序列化
    arr = np.ndarray(data_shape, dtype=np.float64, buffer=shm.buf)

    # ... 并行计算 ...

    shm.close()
    shm.unlink()

3.4 混合模型：asyncio + 进程池

async def hybrid_concurrent(urls, compute_fn, max_io=100, max_cpu=4):
    """混合并发模型：asyncio 处理 I/O，进程池处理 CPU"""
    loop = asyncio.get_event_loop()
    process_pool = ProcessPoolExecutor(max_workers=max_cpu)

    # 第一阶段：异步 I/O 获取数据
    raw_data = await fetch_urls_async(urls, max_concurrent=max_io)

    # 第二阶段：进程池并行处理 CPU 密集计算
    loop_tasks = []
    for url, data in raw_data.items():
        task = loop.run_in_executor(
            process_pool, compute_fn, data
        )
        loop_tasks.append(task)

    results = await asyncio.gather(*loop_tasks)
    process_pool.shutdown(wait=True)
    return results

四、并发模型的性能边界与选型陷阱

Threading 的 GIL 瓶颈：CPU 密集型任务在多线程下不仅没有加速，反而可能更慢——GIL 的获取/释放本身有开销，线程上下文切换也有成本。一个纯 CPU 计算的循环在 4 线程下可能比单线程慢 20%-30%。

Asyncio 的生态割裂：asyncio 要求所有 I/O 操作使用异步库（aiohttp 而非 requests，aiomysql 而非 pymysql）。同步库的阻塞调用会冻结整个事件循环，导致所有协程停顿。在已有同步代码库中引入 asyncio，需要大规模重写 I/O 层。

Multiprocessing 的内存开销：每个子进程有独立的内存空间，启动一个子进程约需 50-100ms 和 30-50MB 内存。频繁创建销毁进程的开销巨大，需要使用进程池。进程间通信（Queue、Pipe）需要序列化数据，大对象的序列化/反序列化开销可能抵消并行收益。

混合模型的调试复杂度：asyncio + 进程池的混合模型，异常处理和调试难度显著增加。进程池中的异常需要通过 Future.result() 重新抛出，traceback 可能丢失。建议在混合模型中使用结构化日志，为每个请求分配唯一 ID，跨 I/O 和 CPU 阶段追踪。

五、总结

Python 并发选型的本质是在"GIL 限制"和"任务特征"之间找到匹配方案。本文的核心结论：纯 I/O 密集选 asyncio（高并发、低开销），I/O + CPU 混合选 asyncio + 进程池，纯 CPU 密集选 multiprocessing。落地时需重点关注三个参数：线程/协程并发度（I/O 建议不超过 1000）、进程池大小（建议等于 CPU 核心数）、任务超时时间（建议 30 秒）。建议通过性能基准测试量化不同模型的实际吞吐和延迟，而非依赖经验判断。