Python 并发编程实战: async/await、多线程与多进程选型指南

问题场景

假设我们需要处理 100 个 URL 的数据抓取任务，每个请求耗时约 0.5 秒。同步方案需要 50 秒，显然不可接受。Python 提供了三种并发方案，该如何选择？

方案一: asyncio (异步 IO)

适合 IO 密集型 任务——网络请求、文件读写、数据库查询。单线程事件循环，资源开销最小。

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=10) as resp:
            data = await resp.text()
            return {"url": url, "status": resp.status, "size": len(data)}
    except Exception as e:
        return {"url": url, "error": str(e)}

async def fetch_all(urls: list[str], concurrency: int = 10):
    semaphore = asyncio.Semaphore(concurrency)

    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch_url(session, url)

    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [bounded_fetch(session, u) for u in urls]
        return await asyncio.gather(*tasks)

async def main():
    urls = [f"https://httpbin.org/delay/0.5?n={i}" for i in range(100)]
    start = time.perf_counter()
    results = await fetch_all(urls, concurrency=20)
    elapsed = time.perf_counter() - start
    print(f"100 requests in {elapsed:.1f}s")  # ~2.5s

asyncio.run(main())

关键点: - Semaphore 控制并发上限，避免打爆目标服务器 - aiohttp 代替 requests——后者是同步库，会阻塞事件循环 - 单线程无 GIL 竞争，适合大量 IO 等待

方案二: threading (多线程)

适合 IO 密集型但无法用 async 的场景——调用不支持 async 的第三方库。

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import time

def fetch_sync(url: str) -> dict:
    try:
        resp = requests.get(url, timeout=10)
        return {"url": url, "status": resp.status_code, "size": len(resp.text)}
    except Exception as e:
        return {"url": url, "error": str(e)}

def fetch_with_threads(urls: list[str], workers: int = 20):
    start = time.perf_counter()
    results = []
    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_map = {executor.submit(fetch_sync, u): u for u in urls}
        for future in as_completed(future_map):
            results.append(future.result())
    elapsed = time.perf_counter() - start
    print(f"100 requests in {elapsed:.1f}s")
    return results

线程安全陷阱:

# 错误: 线程间共享可变状态不加锁
counter = 0

def unsafe_increment():
    global counter
    for _ in range(100_000):
        counter += 1  # 非原子操作!

# 正确: 使用 Lock
from threading import Lock
lock = Lock()
safe_counter = 0

def safe_increment():
    global safe_counter
    for _ in range(100_000):
        with lock:
            safe_counter += 1

方案三: multiprocessing (多进程)

适合 CPU 密集型 任务——图像处理、数据计算、机器学习推理。绕过 GIL，真正利用多核。

from concurrent.futures import ProcessPoolExecutor
import hashlib
import time

def cpu_intensive(n: int) -> dict:
    data = b"Python concurrent programming " * 10_000
    start = time.perf_counter()
    for _ in range(n):
        hashlib.sha256(data).hexdigest()
    elapsed = time.perf_counter() - start
    return {"task": n, "elapsed": f"{elapsed:.2f}s"}

def run_cpu_tasks():
    workloads = [500, 500, 500, 500, 500, 500, 500, 500]
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_intensive, workloads))
    total = time.perf_counter() - start
    print(f"8 tasks / 4 processes: {total:.1f}s")
    return results

if __name__ == "__main__":
    run_cpu_tasks()

选型速查表

核心原则: IO 密集选 async，CPU 密集选多进程，混用场景先 profiling 再优化。