Python 并发与异步

1. 并发基础

1.1 核心概念辨析

概念	说明
并发（Concurrency）	多个任务交替执行，看起来同时（单核也可并发）
并行（Parallelism）	多个任务真正同时执行，依赖多核 CPU
同步（Synchronous）	调用后阻塞等待结果再继续
异步（Asynchronous）	调用后立即返回，结果通过回调 / await 获取
阻塞（Blocking）	当前线程停止等待某个操作（如 I/O）完成
非阻塞（Non-blocking）	操作未就绪时立即返回，不等待

1.2 选择并发方案

任务类型	瓶颈	推荐方案
CPU 密集型	计算时间	多进程（绕过 GIL，利用多核）
I/O 密集型	等待 I/O 时间	多线程（简单）或 异步 I/O（高并发）
高并发 I/O	连接数量	异步 I/O（单线程事件循环，极低开销）

Python 提供三种并发方式：

• 多线程：threading 模块，受 GIL 限制，适合 I/O 密集型
• 多进程：multiprocessing 模块，独立进程，适合 CPU 密集型
• 异步 I/O：asyncio 模块，单线程事件循环，适合高并发 I/O

2. 多线程

2.1 线程基础

import threading

def worker(n):
    print(f"线程 {n} 开始")
    # ... 执行任务 ...
    print(f"线程 {n} 结束")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads: t.start()
for t in threads: t.join()   # 等待所有线程完成

守护线程：主线程退出时自动结束，适合后台任务：

t = threading.Thread(target=background_task, daemon=True)
t.start()
# 主线程结束时 t 会被自动终止

2.2 同步原语

线程共享内存，并发访问共享资源可能导致竞态条件：

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    with lock:        # 自动获取和释放锁
        counter += 1  # 临界区

threads = [threading.Thread(target=increment) for _ in range(1000)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)   # 1000（加锁保证正确性）

常用同步原语：

原语	说明
Lock	互斥锁，同一时刻只有一个线程持有
RLock	可重入锁，同一线程可多次获取（防止死锁）
Semaphore	信号量，控制同时访问资源的线程数量
Event	线程间事件通知，一个线程通知，多个线程等待
Condition	条件变量，结合锁使用，支持 wait / notify
Barrier	栅栏，等待一组线程都到达某点后同时继续

# Event 示例：生产者通知消费者
event = threading.Event()

def producer():
    time.sleep(1)
    data = "ready"
    event.set()   # 通知消费者

def consumer():
    event.wait()  # 阻塞直到 event 被 set
    print("数据就绪，开始处理")

# Semaphore 示例：限制并发数量
semaphore = threading.Semaphore(3)   # 最多 3 个线程同时执行

def limited_task(n):
    with semaphore:
        print(f"任务 {n} 执行中")
        time.sleep(1)

2.3 线程池

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch(url):
    # 模拟 HTTP 请求
    return f"result of {url}"

urls = [f"http://example.com/{i}" for i in range(10)]

with ThreadPoolExecutor(max_workers=4) as executor:
    # 方式一：map（保持顺序，阻塞直到全部完成）
    results = list(executor.map(fetch, urls))

    # 方式二：submit + as_completed（哪个先完成先处理）
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
        except Exception as e:
            print(f"{url} 失败: {e}")

2.4 GIL（全局解释器锁）

CPython 的 GIL 保证同一时刻只有一个线程执行 Python 字节码：

GIL 的本质：保护 CPython 解释器内部数据结构的线程安全
代价：CPU 密集型任务无法利用多核
释放时机：I/O 操作（网络、文件、sleep）会释放 GIL

# I/O 密集型：多线程有效（GIL 在等待 I/O 时释放）
# CPU 密集型：多线程无效，用多进程
import time, threading

def cpu_task():
    total = 0
    for _ in range(10_000_000):
        total += 1

# 单线程耗时 ≈ 多线程耗时（GIL 没有帮助）

Python 3.13+ 的变化：CPython 正在实现"no-GIL"模式（PEP 703），未来多线程可真正并行执行 CPU 密集型任务。

3. 多进程

3.1 进程基础

from multiprocessing import Process
import os

def worker(n):
    print(f"进程 {n}，PID: {os.getpid()}")

if __name__ == "__main__":   # Windows 必须有此保护，防止无限创建子进程
    processes = [Process(target=worker, args=(i,)) for i in range(4)]
    for p in processes: p.start()
    for p in processes: p.join()

进程间不共享内存，每个进程有独立的 Python 解释器，绕开 GIL。

3.2 进程间通信（IPC）

由于进程不共享内存，需要特殊机制传递数据：

from multiprocessing import Queue, Pipe, Manager, Value, Array
import ctypes

# Queue：进程安全的消息队列
def producer(q):
    for i in range(5):
        q.put(i)
    q.put(None)   # 哨兵值，通知消费者结束

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"处理: {item}")

q = Queue()
p1 = Process(target=producer, args=(q,))
p2 = Process(target=consumer, args=(q,))

# Pipe：双向通信管道（比 Queue 更快，但只支持两端）
parent_conn, child_conn = Pipe()

# 共享内存（适合简单数据）
shared_val = Value(ctypes.c_int, 0)   # 共享整数
shared_arr = Array(ctypes.c_double, [1.0, 2.0, 3.0])   # 共享数组

# Manager：共享复杂数据结构（有序列化开销）
with Manager() as manager:
    shared_list = manager.list()
    shared_dict = manager.dict()

3.3 进程池

from multiprocessing import Pool
from concurrent.futures import ProcessPoolExecutor

def compute(x):
    return x ** 2

if __name__ == "__main__":
    # multiprocessing.Pool
    with Pool(processes=4) as pool:
        results = pool.map(compute, range(10))
        # pool.imap：惰性返回，节省内存
        for result in pool.imap(compute, range(1000)):
            process(result)

    # concurrent.futures（接口更统一）
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(compute, range(10)))

选择原则：

• ProcessPoolExecutor：接口与 ThreadPoolExecutor 统一，推荐优先使用
• multiprocessing.Pool：需要 imap/imap_unordered 等高级功能时使用

4. 协程与异步 I/O

4.1 协程的本质

协程是可在执行中暂停和恢复的函数，由程序自身控制切换：

对比维度	线程	协程
调度方式	操作系统（抢占式）	事件循环（协作式，主动让出）
切换开销	有系统调用开销（上下文切换）	极低（用户态切换）
内存占用	每个线程约 1MB 栈空间	每个协程极少（KB 级别）
并发数量	数千	数万甚至数十万
适合场景	I/O 密集 + 简单逻辑	高并发 I/O（网络服务、爬虫）

4.2 async / await

import asyncio

async def fetch_data(name, delay):
    print(f"开始获取 {name}")
    await asyncio.sleep(delay)   # 等待期间让出控制权，可执行其他协程
    print(f"完成获取 {name}")
    return f"data: {name}"

async def main():
    # 串行执行（总耗时 = 各任务之和）
    # r1 = await fetch_data("A", 1)
    # r2 = await fetch_data("B", 1)

    # 并发执行（总耗时 = 最长任务时间）
    results = await asyncio.gather(
        fetch_data("A", 1),
        fetch_data("B", 0.5),
        fetch_data("C", 1.5),
    )
    print(results)   # ['data: A', 'data: B', 'data: C']

asyncio.run(main())

await 只能在 async def 内使用，表示"等待一个可等待对象，等待期间让出控制权给事件循环"。

4.3 事件循环

事件循环是异步 I/O 的核心调度器：

事件循环运行流程：
1. 获取就绪的任务（I/O 完成、定时器到期等）
2. 依次执行就绪任务，直到每个任务主动 await（让出控制权）
3. 轮询等待更多 I/O 事件
4. 重复 1-3，直到所有任务完成

# 推荐方式（Python 3.7+）
asyncio.run(main())   # 创建新事件循环，运行完后关闭

# 底层访问（框架开发、Jupyter 等特殊场景）
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

4.4 Task 与 Future

Future：表示一个异步操作的最终结果，是底层的"占位符"对象。
Task：对协程的封装，是 Future 的子类，会被调度到事件循环中执行。

async def main():
    # create_task 立即将协程提交给事件循环调度
    task1 = asyncio.create_task(fetch_data("A", 1))
    task2 = asyncio.create_task(fetch_data("B", 0.5))

    # 等待各自完成
    result1 = await task1
    result2 = await task2

    # 等待多个任务
    results = await asyncio.gather(task1, task2)

    # 设置超时（Python 3.11+）
    try:
        result = await asyncio.wait_for(fetch_data("C", 5), timeout=2.0)
    except asyncio.TimeoutError:
        print("超时！")

任务取消：

async def main():
    task = asyncio.create_task(long_running())
    await asyncio.sleep(1)
    task.cancel()   # 发送取消信号

    try:
        await task
    except asyncio.CancelledError:
        print("任务已取消")

asyncio.gather vs asyncio.wait：

# gather：等待全部完成，异常会传播
results = await asyncio.gather(*tasks, return_exceptions=True)

# wait：可以按完成顺序处理，支持 FIRST_COMPLETED 等策略
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for task in done:
    print(task.result())

4.5 异步 I/O

真正的异步 I/O 需要使用支持异步的库（普通的 requests.get、open() 会阻塞事件循环）：

import aiohttp
import aiofiles

# 异步 HTTP（aiohttp）
async def fetch_http(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# 并发抓取多个 URL
async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_http(url) for url in urls]
        return await asyncio.gather(*tasks)

# 异步文件 I/O（aiofiles）
async def read_file(path):
    async with aiofiles.open(path, 'r') as f:
        return await f.read()

# 异步数据库（asyncpg / aiomysql / aiosqlite）
import asyncpg
async def query_db():
    conn = await asyncpg.connect("postgresql://...")
    rows = await conn.fetch("SELECT * FROM users")
    await conn.close()
    return rows

在协程中调用阻塞代码（run_in_executor）：

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

def blocking_io():
    time.sleep(1)   # 阻塞操作（无法直接 await）
    return "done"

async def main():
    loop = asyncio.get_running_loop()
    # 在线程池中运行阻塞函数，不阻塞事件循环
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

    # 使用自定义线程池
    with ThreadPoolExecutor(max_workers=4) as pool:
        result = await loop.run_in_executor(pool, blocking_io)

4.6 并发控制

asyncio.Semaphore：限制并发数量（常用于爬虫速率限制）：

async def fetch_with_limit(session, url, semaphore):
    async with semaphore:   # 最多 5 个并发请求
        async with session.get(url) as response:
            return await response.text()

async def main():
    semaphore = asyncio.Semaphore(5)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)

asyncio.Queue：生产者-消费者模式：

async def producer(queue):
    for i in range(10):
        await queue.put(i)
        print(f"生产: {i}")
    await queue.put(None)   # 哨兵值

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)   # 传递给其他消费者
            break
        await asyncio.sleep(0.1)   # 模拟处理
        print(f"消费者 {name} 处理: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=3)   # 最多缓存 3 个
    await asyncio.gather(
        producer(queue),
        consumer(queue, "A"),
        consumer(queue, "B"),
    )

4.7 异步上下文管理器

实现 __aenter__ 和 __aexit__ 协议，配合 async with 使用：

class AsyncDBConnection:
    async def __aenter__(self):
        self.conn = await asyncpg.connect("postgresql://...")
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()
        return False

async with AsyncDBConnection() as conn:
    await conn.execute("SELECT 1")

简洁写法（用 asynccontextmanager）：

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    # 启动时执行
    conn = await asyncpg.connect("postgresql://...")
    app.db = conn
    yield
    # 关闭时执行
    await conn.close()

4.8 异步迭代器与异步生成器

异步迭代器：实现 __aiter__ 和 __anext__，配合 async for 使用。

异步生成器（更简洁）：

async def paginate(api, page_size=100):
    page = 0
    while True:
        data = await api.fetch(page=page, size=page_size)
        if not data:
            break
        for item in data:
            yield item   # async generator
        page += 1

async def main():
    async for item in paginate(api):
        process(item)

5. 并发模型对比与选型

场景                    推荐方案
─────────────────────────────────────────────
CPU 密集（数学计算等）  → 多进程（ProcessPoolExecutor）
I/O 密集（有限并发）   → 多线程（ThreadPoolExecutor）
I/O 密集（高并发）     → 异步 I/O（asyncio + aiohttp）
混合（CPU + I/O）      → 异步 I/O + run_in_executor

常见误区：

• 多线程 ≠ 并行：CPython 的 GIL 使 CPU 密集型多线程无法真正并行
• 异步 ≠ 多核：asyncio 是单线程，不利用多核（可配合 ProcessPoolExecutor）
• 异步库不可混用阻塞调用：在 async 函数中调用 requests.get 会阻塞整个事件循环