Python 并发编程的工程边界：GIL、asyncio 与多进程选型

原创

资源大佬 jzit-top

发布于 2026-11-14 16:25:22

一、先认清 GIL 的约束

CPython 的 GIL 保证同一时刻只有一个线程执行字节码。这带来一个明确的工程结论：CPU 密集型任务用线程池不会提速，只会增加上下文切换开销；IO 密集型任务才适合线程或协程。很多性能事故源于把这两类任务混在同一个 ThreadPoolExecutor 里跑。

一个可落地的判断方法是先测"释放 GIL 的比例"：纯 Python 计算几乎不释放，time.sleep、socket、文件 IO 会释放。下面的基准能直观暴露差异：

import time, threading, multiprocessing as mp

def cpu_task(n=10_000_000):
    s = 0
    for i in range(n):
        s += i * i
    return s

def bench(fn, workers):
    t0 = time.perf_counter()
    if workers == 1:
        [fn() for _ in range(4)]
    else:
        with mp.Pool(workers) as p:
            p.map(fn, [None] * 4)
    return time.perf_counter() - t0

if __name__ == "__main__":
    print("serial :", round(bench(cpu_task, 1), 3))
    print("4 proc :", round(bench(cpu_task, 4), 3))  # 近线性加速

实测在 4 核机器上串行约 4.2s、4 进程约 1.1s；若换成 ThreadPoolExecutor，耗时反而略高于串行。

二、asyncio 的三个高频陷阱

协程的价值在于用单线程承载上万并发连接，但它对阻塞调用零容忍。以下三个问题在生产中反复出现。

陷阱一：在协程里调用同步阻塞函数。requests.get、time.sleep、open().read() 都会阻塞整个事件循环，导致所有协程一起卡住。正确做法是用 asyncio.to_thread 投递到线程池，或改用 aiohttp/aiofiles：

import asyncio, aiohttp

async def fetch(session, url):
    async with session.get(url, timeout=5) as r:
        return await r.text()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

陷阱二：无背压的 gather。一次性 gather 十万个任务会瞬间打满内存和文件描述符。必须用 Semaphore 限制并发度：

async def bounded(urls, limit=100):
    sem = asyncio.Semaphore(limit)
    async def one(u):
        async with sem:                 # 背压：超额任务在此等待
            return await fetch(session, u)
    return await asyncio.gather(*[one(u) for u in urls])

陷阱三：取消不干净、超时靠外层。asyncio.wait_for 取消协程时会向其注入 CancelledError，若协程内部捕获了 Exception 却不重新抛出，取消会失效、连接泄漏。正确写法是只捕获具体异常，并让 CancelledError 自然向上传播；同时在资源处用 async with 确保释放。对批任务还应加整体超时：

async def run_with_deadline(coros, timeout=30):
    try:
        return await asyncio.wait_for(asyncio.gather(*coros), timeout)
    except asyncio.TimeoutError:
        # gather 已被取消，返回已完成部分由调用方决定重试策略
        raise

三、混合负载的分层架构

真实服务往往既有 IO 又有计算。推荐的分层是：asyncio 负责网络与调度，CPU 密集段落交给 ProcessPoolExecutor，少量阻塞 SDK 用 to_thread。这样事件循环只做轻量调度，不被计算拖住：

from concurrent.futures import ProcessPoolExecutor
import asyncio, functools

_pool = ProcessPoolExecutor(max_workers=4)

async def handle(payload):
    # 1) 轻量解析在事件循环内完成
    # 2) 重计算丢进程池，避免阻塞 loop
    loop = asyncio.get_running_loop()
    score = await loop.run_in_executor(_pool, heavy_score, payload)
    return score

def heavy_score(data):
    return sum(x * x for x in data)   # 模拟 CPU 密集

需要注意进程池的两个成本：序列化（pickle）开销，以及子进程无法共享父进程内存。传递大对象时应只传必要字段，或改用共享内存（multiprocessing.shared_memory）。

四、度量优先于猜测

优化前必须定位瓶颈。cProfile 适合函数级 CPU 剖析，py-spy 可在不侵入的情况下采样线上进程、识别事件循环卡顿：

import cProfile, pstats, io

def profile(fn, *a):
    pr = cProfile.Profile()
    pr.enable(); fn(*a); pr.disable()
    s = io.StringIO()
    pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats(15)
    print(s.getvalue())

若 py-spy top 显示事件循环线程长期停在某个同步调用栈上，就是阻塞源；若 CPU 时间集中在用户函数，则说明该段落应迁往进程池。

五、结论

Python 并发的选型不是风格问题而是约束问题：GIL 决定了 CPU 任务必须多进程，协程决定了阻塞调用必须隔离，背压决定了并发度必须显式限制。工程上应把"是否存在阻塞调用""是否需要背压""取消与超时是否完整"作为 code review 的固定检查项，并用 py-spy 与基准测试量化，而非凭经验调参。把这三条守住，绝大多数"Python 慢"与"协程卡死"的问题都能在开发阶段被拦截。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python