python

Python 并发与性能：GIL、asyncio 与多进程的工程实践

深度解析 Python 并发模型：GIL 内部机制与绕过策略、asyncio 事件循环与 Task 调度、async/await 最佳实践、multiprocessing 多核利用、concurrent.futures 线程池/进程池、cProfile/py-spy/flamegraph 性能分析。附带可运行代码与工作负载选型决策树。

spcent 5 分钟阅读 2144 字

本文假设 Python 3.11+。所有代码可直接复制运行。

1. GIL（全局解释器锁）的真相

1.1 什么是 GIL

CPython 的 GIL（Global Interpreter Lock） 是一个互斥锁，确保在任意时刻只有一个线程在执行 Python 字节码。

import sys

print(f"GIL enabled: {sys._is_gil_enabled()}")  # Python 3.13+ 可查询

GIL 保护的对象：

  • CPython 内存分配（PyMem_Malloc）
  • 引用计数增减
  • 垃圾回收（GC）

例外： 执行 C 扩展（如 NumPy 矩阵运算）时，C 代码可以释放 GIL。

1.2 GIL 对性能的实际影响

import threading
import time

def cpu_bound(n: int) -> int:
    """纯 CPU 计算，GIL 串行化"""
    count = 0
    for i in range(n):
        count += i ** 2
    return count

# 单线程
start = time.perf_counter()
cpu_bound(10_000_000)
print(f"Single: {time.perf_counter() - start:.2f}s")

# 多线程（ expect 无加速，甚至更慢）
threads = [threading.Thread(target=cpu_bound, args=(5_000_000,)) for _ in range(2)]
start = time.perf_counter()
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Multi-thread: {time.perf_counter() - start:.2f}s")

输出（典型）：

Single: 0.85s
Multi-thread: 1.05s   # 更慢！因为线程切换开销 + GIL 竞争

1.3 GIL 不限制的场景

场景能否并行原因
I/O 等待（网络、文件）✅ 可以read() 释放 GIL，线程可切换
NumPy/SciPy 批量计算✅ 可以C 代码释放 GIL
纯 Python CPU 密集❌ 不行GIL 串行化

2. asyncio：单线程并发的首选

2.1 核心概念：事件循环

import asyncio

async def say_hello():
    await asyncio.sleep(1)
    print("Hello")

asyncio.run(say_hello())  # 创建事件循环 → 运行协程 → 关闭循环

asyncio 架构：

┌─────────────────┐
│   Event Loop    │ ← 单线程调度核心
├─────────────────┤
│   Selector      │ ← epoll/kqueue/IOCP
├─────────────────┤
│   Task Queue    │ ← 就绪任务 FIFO/优先级
├─────────────────┤
│   Callbacks     │ ← Future 完成回调
└─────────────────┘

2.2 Task、Future 与协程的关系

import asyncio

async def fetch_data(url: str) -> dict:
    await asyncio.sleep(0.5)  # 模拟 I/O
    return {"url": url, "status": 200}

async def main():
    # 创建任务（立即注册到事件循环）
    task1 = asyncio.create_task(fetch_data("https://api.a.com"))
    task2 = asyncio.create_task(fetch_data("https://api.b.com"))
    
    # 并发等待
    results = await asyncio.gather(task1, task2)
    print(results)

asyncio.run(main())

区别：

概念说明
Coroutineasync def 定义的函数对象，惰性执行
Task包装协程，注册到事件循环，可被取消/等待
Future低层回调容器，通常由库创建

2.3 并发模式速查

# 1. 并发收集（全部完成）
results = await asyncio.gather(*coros)

# 2. 并发收集（任一完成）
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

# 3. 带超时
try:
    result = await asyncio.wait_for(slow_coro(), timeout=5.0)
except asyncio.TimeoutError:
    ...

# 4. 带 semaphore 限制并发数
sem = asyncio.Semaphore(10)
async def bounded_fetch(url):
    async with sem:
        return await fetch(url)

# 5. 后台任务 + 优雅取消
async def background_worker(queue: asyncio.Queue):
    try:
        while True:
            item = await queue.get()
            await process(item)
    except asyncio.CancelledError:
        print("Worker shutting down...")
        raise

2.4 asyncio 最佳实践

✅ 推荐：

  • 所有 I/O 函数都声明为 async
  • asyncio.create_task() 而非 asyncio.ensure_future()
  • 使用 asyncio.gather() 并发 I/O
  • 顶层入口点用 asyncio.run()（3.7+）

❌ 避免：

  • 在 async def 中调用同步 I/O（阻塞事件循环）
  • 在协程中使用 time.sleep()（需要用 asyncio.sleep()）
  • 不加限制地并发（可能触发 Too many open files）

同步 I/O 转异步：

import asyncio

# 错误：阻塞事件循环！
# requests.get(url)

# 正确：使用 aiohttp 或 loop.run_in_executor
import aiohttp

async def fetch_async(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as resp:
        return await resp.json()

# 无法修改的同步库 → 线程池
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, sync_func, arg1, arg2)

3. multiprocessing：绕过 GIL 的多核计算

3.1 基本使用

from multiprocessing import Pool
import time

def cpu_intensive(n: int) -> int:
    """CPU 密集型计算"""
    return sum(i ** 2 for i in range(n))

if __name__ == "__main__":
    inputs = [5_000_000] * 4
    
    start = time.perf_counter()
    with Pool(processes=4) as pool:
        results = pool.map(cpu_intensive, inputs)
    print(f"Multiprocessing: {time.perf_counter() - start:.2f}s")

3.2 multiprocessing 开销分析

开销项说明
fork/spawn子进程创建（Linux fork 快，Windows spawn 慢）
Pickling参数/结果序列化（大数据量成本高）
内存复制写时复制（COW），共享大数组用 shared_memory

使用 shared_memory 避免复制：

from multiprocessing import Process, shared_memory
import numpy as np

def worker(shm_name, shape, dtype):
    shm = shared_memory.SharedMemory(name=shm_name)
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    arr += 1  # 原地修改
    shm.close()

if __name__ == "__main__":
    arr = np.zeros((10000, 10000), dtype=np.float32)
    shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
    shared_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
    shared_arr[:] = arr[:]
    
    p = Process(target=worker, args=(shm.name, arr.shape, arr.dtype))
    p.start()
    p.join()
    
    print(shared_arr.sum())  # 修改已生效
    shm.close()
    shm.unlink()

3.3 concurrent.futures：统一 API

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def task(n):
    time.sleep(1)
    return n * n

# I/O 密集型 → 线程池
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(task, range(10)))

# CPU 密集型 → 进程池
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(cpu_intensive, [1_000_000] * 4))

3.4 选型决策树

工作负载类型？
├── I/O 密集型（网络、文件、DB）
│   └── asyncio（单线程最高效）
│   └── 或 ThreadPoolExecutor（简单同步代码）
├── CPU 密集型（计算、数据处理）
│   └── multiprocessing / ProcessPoolExecutor
│   └── 或 numpy（C 代码释放 GIL）
└── 混合负载
    └── asyncio + ProcessPoolExecutor（主协程 + 子进程池）

4. 性能分析工具链

4.1 cProfile：标准库分析器

python -m cProfile -s cumulative script.py

代码内使用：

import cProfile
import pstats

profiler = cProfile.Profile()
profiler.enable()
# ... 你的代码 ...
profiler.disable()

stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(20)  # Top 20

输出解读：

列含义
ncalls调用次数
tottime函数自身执行时间（不含子调用）
cumtime累计时间（含子调用）
percall每次调用平均

4.2 py-spy：采样分析器（推荐）

# 安装
pip install py-spy

# 实时查看
py-spy top --pid <PID>

# 生成火焰图
py-spy record -o profile.svg --python --pid <PID>

特点：

  • 无需修改代码
  • 对生产环境影响极小（<5% CPU）
  • 支持火焰图可视化

4.3 其他工具

工具场景命令
line_profiler逐行耗时分析@profile 装饰
memory_profiler内存分析@profile + mprof run
scaleneCPU + 内存 + Python/C 混合scalene script.py
pytest-benchmark基准测试pytest --benchmark-only

4.4 火焰图解读

火焰图（Flame Graph）是性能分析的可视化神器：

# 生成
py-spy record -o flame.svg --python -- script.py

# 解读技巧：
# - 横轴 = 时间占比（不是时间线）
# - 纵轴 = 调用栈深度
# - 颜色 = 随机（非热度），看宽度！
# - 底部宽 = 大头，优先优化

5. 实战：高并发爬虫

import asyncio
import aiohttp
from typing import List

async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            return {"url": url, "status": resp.status}
    except Exception as e:
        return {"url": url, "error": str(e)}

async def crawl(urls: List[str], max_concurrent: int = 50) -> List[dict]:
    semaphore = asyncio.Semaphore(max_concurrent)
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)
    
    async with aiohttp.ClientSession(connector=connector) as session:
        async def bounded_fetch(url: str) -> dict:
            async with semaphore:
                return await fetch(session, url)
        
        return await asyncio.gather(*[bounded_fetch(u) for u in urls])

# 运行
urls = [f"https://httpbin.org/get?i={i}" for i in range(1000)]
results = asyncio.run(crawl(urls))
print(f"Success: {sum(1 for r in results if 'status' in r)}")

6. 常见陷阱与解法

陷阱现象解法
协程里调用同步 I/O事件循环阻塞，并发数归零用 run_in_executor 或改异步库
忘记 await得到 Coroutine 对象而非结果类型检查 + lint
异常吞没gather() 中一个异常，其余取消return_exceptions=True
内存泄漏Task 引用循环weakref + 显式 cancel()
进程死锁Pool 中调用了 Pool只在主进程创建 Pool

延伸阅读

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

Python 部署与分发：Docker、PyPI 发布与可复现环境 Python 项目从开发到生产的完整部署路径：Docker 多阶段构建与镜像优化、uvicorn/gunicorn 服务器配置、pyinstaller/uv 打包独立可执行文件、PyPI 包发布流程、Nix 可复现环境。附带 Dockerfile 模板和 GitHub Actions 发布流水线。 Python 测试与质量工程：pytest、mock 与覆盖率实战 Python 测试金字塔完整实践：pytest 核心（fixture/parametrize/monkeypatch）、unittest.mock/patch、Monkeypatch、覆盖率 pytest-cov、类型测试、CI 集成策略与 doctest。覆盖从单元测试到集成测试的完整工程方案。 Python 现代工具链：uv + ruff + mypy 全链路工程实践 Python 工具链现代化完整指南：uv（极速包管理+虚拟环境+Python 安装）、ruff（lint+format 一体化）、mypy/pyright 类型检查、pipx 工具安装、 hatch/poetry/pdm 项目管理、从 pip 到 uv 的迁移路径。附带 pyproject.toml 完整配置模板。