跳至主要內容

12 并发、并行与异步

大约 3 分钟编程语言Python并发并行asyncio

12 并发、并行与异步

先判断瓶颈是 I/O 等待、CPU 计算还是外部容量。并发不是默认优化，它会增加取消、超时、顺序、限流和状态一致性的复杂度。

1. 模型选择

场景优先方案
少量线程式阻塞 I/Othreading
批量阻塞 I/Oconcurrent.futures.ThreadPoolExecutor
大量异步网络 I/Oasyncio
CPU 密集任务ProcessPoolExecutor 或原生扩展
多个独立进程multiprocessing
跨主机并发消息队列或编排系统

线程受 GIL 影响，但 I/O 等待期间仍可调度。进程绕过 GIL 的代价是序列化、启动成本和共享状态困难。异步要求调用链使用异步接口。

2. 线程池

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

def fetch(name: str) -> str:
    sleep(0.05)
    return f"{name}:ok"

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch, name): name for name in ["a", "b", "c"]}
    for future in as_completed(futures):
        name = futures[future]
        try:
            print(name, future.result())
        except Exception as exc:
            print(name, type(exc).__name__, exc)

线程池必须有上限，并与下游承载能力匹配。收集结果时要处理单个任务异常，避免一个失败吞掉整批结果。

3. 进程池

from concurrent.futures import ProcessPoolExecutor

def square(value: int) -> int:
    return value * value

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        print(list(executor.map(square, range(8))))

传给进程的参数和返回值必须可序列化。不要依赖全局可变状态；Windows 使用 spawn 启动方式，入口必须受到 if __name__ == "__main__" 保护。

4. asyncio 基础

import asyncio

async def fetch(name: str) -> str:
    await asyncio.sleep(0.05)
    return f"{name}:ok"

async def main() -> None:
    results = await asyncio.gather(*(fetch(name) for name in ["a", "b", "c"]))
    print(results)

asyncio.run(main())

异步函数只有被某段事件循环调度后才执行。同步阻塞调用会卡住整个事件循环；在异步服务中调用遗留阻塞库时，应使用线程池或其他隔离边界。

5. 超时与取消

import asyncio

async def slow_operation() -> None:
    await asyncio.sleep(10)

async def run() -> None:
    try:
        async with asyncio.timeout(1):
            await slow_operation()
    except TimeoutError:
        print("operation timed out")

asyncio.run(run())

asyncio.timeout 需要 Python 3.11 或更高版本。取消是正常控制流，资源清理和状态回滚必须能被取消，并在他方取消时继续传播 CancelledError。

6. 有界并发

import asyncio

async def worker(semaphore: asyncio.Semaphore, name: str) -> str:
    async with semaphore:
        await asyncio.sleep(0.01)
        return name

async def main() -> None:
    semaphore = asyncio.Semaphore(10)
    results = await asyncio.gather(*(worker(semaphore, str(i)) for i in range(100)))
    print(len(results))

asyncio.run(main())

不要因为异步很容易创建大量任务就无限并发。真实约束通常是连接池、远端限流、文件描述符、内存和数据库容量。

7. 共享状态

多线程共享数据时必须使用锁或不共享可变状态：

import threading

class Counter:
    def __init__(self) -> None:
        self._value = 0
        self._lock = threading.Lock()

    def increment(self) -> int:
        with self._lock:
            self._value += 1
            return self._value

优先让线程只处理独立任务，结果通过队列或 future 汇总。锁范围保持最小，锁内不要执行网络请求和耗时计算。

8. 队列和背压

from queue import Queue

queue: Queue[str] = Queue(maxsize=100)
queue.put("event")
print(queue.get())

有界队列提供背压，防止生产速度远超消费速度而耗尽内存。队列消费需要停机信号、异常处理和可观察的积压指标。

9. 并发检查清单

  • 是否设置了连接数、任务数和队列上限。
  • 是否设置超时、总预算和取消路径。
  • 重试是否有随机抖动和幂等保证。
  • 是否记录并发量、排队时间、失败类型和耗时。
  • 程序停机时是否等待、取消并清理资源。
  • 是否在真实负载下验证内存、连接和远端限流。

官方参考