并发编程完全指南

官方文档：https://docs.python.org/3/library/concurrency.html
适用版本：Python 3.12（2026-05-07 核实）

Python 提供三种主要并发模型：基于线程的 threading、基于进程的 multiprocessing 和基于协程的 asyncio。此外，concurrent.futures 提供了统一的高层接口。选择哪种模型取决于任务类型。

并发模型选择指南

threading 模块

GIL（全局解释器锁）

CPython 的 GIL 确保同一时刻只有一个线程执行 Python 字节码。因此，对于纯 Python 的 CPU 密集型代码，多线程不能带来并行加速。但对于 IO 密集型任务，线程在等待 IO 时会释放 GIL，其他线程可以运行。

C 扩展（如 NumPy 的计算部分）在执行期间可以主动释放 GIL，实现真正的并行。

Thread 基础

import threading
import time

def worker(name: str, duration: float):
    print(f"线程 {name} 开始")
    time.sleep(duration)
    print(f"线程 {name} 结束")

# 创建线程
t1 = threading.Thread(target=worker, args=("A", 1.0))
t2 = threading.Thread(target=worker, args=("B", 0.5), daemon=True)

# 启动
t1.start()
t2.start()

# 等待完成
t1.join()
t2.join(timeout=2.0)  # 最多等待 2 秒

print(f"t1 是否存活: {t1.is_alive()}")

threading.Thread 的构造参数：

join 方法的参数：

守护线程：daemon=True 的线程在主线程退出时自动终止，不会阻止程序退出。适合后台监控任务，但不适合需要完成清理工作的任务。

继承 Thread 类

import threading

class WorkerThread(threading.Thread):
    def __init__(self, data: list, result: list):
        super().__init__()
        self.data = data
        self.result = result

    def run(self):
        # 重写 run() 方法，不要直接调用 run()，要调用 start()
        self.result.extend(x * 2 for x in self.data)


results = []
t = WorkerThread([1, 2, 3], results)
t.start()
t.join()
print(results)  # [2, 4, 6]

Lock（互斥锁）

Lock 保证同一时刻只有一个线程访问共享资源，防止竞态条件。

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    with lock:           # 推荐：用 with 语句自动释放
        counter += 1

# 等价的手动写法
def increment_manual():
    global counter
    lock.acquire()
    try:
        counter += 1
    finally:
        lock.release()

Lock.acquire 的参数：

RLock（可重入锁）

RLock 允许同一线程多次获取锁，避免死锁。适用于递归函数或需要在持有锁时调用其他也需要锁的方法的场景。

import threading

class SafeCounter:
    def __init__(self):
        self._lock = threading.RLock()
        self._value = 0

    def increment(self):
        with self._lock:
            self._value += 1
            self._notify()  # 内部调用也需要锁

    def _notify(self):
        with self._lock:  # RLock 允许同一线程再次获取，Lock 会死锁
            print(f"当前值: {self._value}")

Semaphore（信号量）

Semaphore 限制同时访问资源的线程数量，适用于限制并发连接数等场景。

import threading
import time

# 最多允许 3 个线程同时执行
semaphore = threading.Semaphore(3)

def access_resource(thread_id: int):
    with semaphore:
        print(f"线程 {thread_id} 获得访问权")
        time.sleep(1)
        print(f"线程 {thread_id} 释放访问权")

threads = [threading.Thread(target=access_resource, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

threading.Semaphore 的构造参数：

Event（事件）

Event 用于线程间的简单信号通知，一个线程等待某个条件，另一个线程触发它。

import threading
import time

event = threading.Event()

def waiter():
    print("等待事件...")
    event.wait()         # 阻塞直到 event 被 set
    print("事件触发，继续执行")

def setter():
    time.sleep(2)
    print("触发事件")
    event.set()

threading.Thread(target=waiter).start()
threading.Thread(target=setter).start()

Event 的主要方法：

wait 的参数：

Condition（条件变量）

Condition 在 Lock 的基础上增加了等待/通知机制，适用于生产者-消费者模式。

import threading
from collections import deque

class BoundedQueue:
    def __init__(self, maxsize: int):
        self._queue = deque()
        self._maxsize = maxsize
        self._cond = threading.Condition()

    def put(self, item):
        with self._cond:
            while len(self._queue) >= self._maxsize:
                self._cond.wait()   # 队列满，等待消费者消费
            self._queue.append(item)
            self._cond.notify_all()  # 通知等待的消费者

    def get(self):
        with self._cond:
            while not self._queue:
                self._cond.wait()   # 队列空，等待生产者生产
            item = self._queue.popleft()
            self._cond.notify_all()  # 通知等待的生产者
            return item

Condition.wait 的参数：

threading.local()（线程本地存储）

threading.local() 为每个线程提供独立的数据存储空间，常用于存储数据库连接、请求上下文等线程私有状态。

import threading

# 创建线程本地存储对象
local_data = threading.local()

def worker(value: int):
    local_data.value = value      # 每个线程有独立的 .value 属性
    import time
    time.sleep(0.1)
    print(f"线程 {threading.current_thread().name}: {local_data.value}")

threads = [threading.Thread(target=worker, args=(i,), name=f"T{i}") for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
# 每个线程打印自己的值，不会互相干扰

multiprocessing 模块

multiprocessing 通过创建子进程来绕过 GIL，每个进程有独立的内存空间，适合 CPU 密集型任务。

Process 基础

from multiprocessing import Process
import os

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

if __name__ == "__main__":
    # 重要：必须在 if __name__ == "__main__" 保护下创建进程
    p = Process(target=worker, args=("子进程",))
    p.start()
    p.join()
    print(f"子进程退出码: {p.exitcode}")

Process 的构造参数：

join 方法的参数：

Pool（进程池）

Pool 维护一组工作进程，适合批量任务处理。

from multiprocessing import Pool

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

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        # map：阻塞直到所有结果完成，返回有序列表
        results = pool.map(square, range(10))
        print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

        # starmap：参数是元组列表，解包后传给函数
        pairs = [(1, 2), (3, 4), (5, 6)]
        results = pool.starmap(lambda a, b: a + b, pairs)

        # imap：返回迭代器，按输入顺序产出结果（惰性）
        for result in pool.imap(square, range(10)):
            print(result)

        # imap_unordered：返回迭代器，哪个先完成先产出（乱序）
        for result in pool.imap_unordered(square, range(10)):
            print(result)

Pool 的构造参数：

map 方法的参数：

starmap 方法的参数：

Queue 和 Pipe（进程间通信）

from multiprocessing import Process, Queue, Pipe

# Queue：多进程安全的队列
def producer(q: Queue):
    for i in range(5):
        q.put(i)
    q.put(None)  # 发送结束信号

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

if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()


# Pipe：双端管道，比 Queue 快但只支持两个进程
def sender(conn):
    conn.send([1, 2, 3])
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=sender, args=(child_conn,))
    p.start()
    print(parent_conn.recv())  # [1, 2, 3]
    p.join()

Queue 的构造参数：

Pipe 的参数：

Manager（共享对象）

Manager 提供进程间共享的 Python 对象（list、dict、Namespace 等），通过代理对象实现，有一定性能开销。

from multiprocessing import Process, Manager

def worker(shared_dict: dict, key: str, value: int):
    shared_dict[key] = value

if __name__ == "__main__":
    with Manager() as manager:
        shared = manager.dict()
        processes = [
            Process(target=worker, args=(shared, f"key{i}", i))
            for i in range(5)
        ]
        for p in processes:
            p.start()
        for p in processes:
            p.join()
        print(dict(shared))

共享内存 Value 和 Array

Value 和 Array 使用底层共享内存，比 Manager 快，但只支持 C 类型数据。

from multiprocessing import Process, Value, Array
import ctypes

def increment(counter: Value, lock):
    for _ in range(1000):
        with lock:
            counter.value += 1

if __name__ == "__main__":
    from multiprocessing import Lock
    counter = Value(ctypes.c_int, 0)   # 类型码 'i' 也可以
    lock = Lock()

    processes = [Process(target=increment, args=(counter, lock)) for _ in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(f"最终值: {counter.value}")  # 应该是 4000


# Array 示例
arr = Array(ctypes.c_double, [1.0, 2.0, 3.0])

Value 的构造参数：

Array 的构造参数：

concurrent.futures 模块

concurrent.futures 提供统一的高层接口，屏蔽了线程和进程的底层差异，是处理并发任务的推荐方式。

ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request

def fetch(url: str) -> str:
    with urllib.request.urlopen(url, timeout=5) as resp:
        return resp.read().decode()

urls = [
    "https://example.com",
    "https://httpbin.org/get",
]

with ThreadPoolExecutor(max_workers=5) as executor:
    # submit：提交单个任务，返回 Future 对象
    future = executor.submit(fetch, "https://example.com")
    result = future.result(timeout=10)

    # map：类似内置 map，按顺序返回结果
    results = list(executor.map(fetch, urls, timeout=10))

    # as_completed：哪个先完成先处理
    futures = {executor.submit(fetch, url): url for url in urls}
    for fut in as_completed(futures, timeout=30):
        url = futures[fut]
        try:
            data = fut.result()
            print(f"{url}: {len(data)} 字节")
        except Exception as e:
            print(f"{url}: 请求失败 {e}")

ThreadPoolExecutor 的构造参数：

ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n: int) -> int:
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_heavy, [1_000_000] * 8))
        print(results)

ProcessPoolExecutor 的构造参数：

submit 方法

返回 Future 对象。

map 方法

as_completed 函数

返回一个迭代器，按完成顺序产出 Future 对象。

wait 函数

from concurrent.futures import wait, FIRST_COMPLETED, ALL_COMPLETED, FIRST_EXCEPTION

futures = [executor.submit(task, i) for i in range(10)]

# 等待所有完成
done, not_done = wait(futures, timeout=60, return_when=ALL_COMPLETED)

# 等待第一个完成
done, not_done = wait(futures, return_when=FIRST_COMPLETED)

# 等待第一个异常
done, not_done = wait(futures, return_when=FIRST_EXCEPTION)

wait 函数的参数：

返回 (done: set, not_done: set) 的具名元组。

Future 对象的方法

from concurrent.futures import Future

future: Future = executor.submit(some_function)

future.result(timeout=10)     # 获取结果，超时抛 TimeoutError，函数异常则重新抛出
future.exception(timeout=10)  # 获取异常（无异常返回 None）
future.cancel()               # 取消任务（若已开始则失败，返回 False）
future.cancelled()            # 是否已取消
future.running()              # 是否正在运行
future.done()                 # 是否已完成（完成、取消、异常都算）
future.add_done_callback(fn)  # 完成时调用 fn(future)

result 方法的参数：

exception 方法的参数：

add_done_callback 方法的参数：

ProcessPoolExecutor vs multiprocessing.Pool 对比

最佳实践

1. 使用 context manager 管理 Executor

from concurrent.futures import ThreadPoolExecutor

# 用 with 语句确保所有线程完成后再继续
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(task, i) for i in range(100)]
# with 块退出时自动调用 shutdown(wait=True)

2. 正确处理异常

from concurrent.futures import ThreadPoolExecutor, as_completed

def risky_task(x: int) -> int:
    if x == 5:
        raise ValueError(f"不接受 {x}")
    return x * 2

with ThreadPoolExecutor() as executor:
    futures = {executor.submit(risky_task, i): i for i in range(10)}
    for future in as_completed(futures):
        input_val = futures[future]
        try:
            result = future.result()
            print(f"任务 {input_val} -> {result}")
        except ValueError as e:
            print(f"任务 {input_val} 失败: {e}")

3. 使用 initializer 共享昂贵资源

from concurrent.futures import ProcessPoolExecutor
import threading

# 线程池：用 threading.local() 实现每线程一个连接
thread_local = threading.local()

def init_connection(db_url: str):
    thread_local.conn = create_connection(db_url)

def process_item(item_id: int) -> dict:
    conn = thread_local.conn   # 每个线程使用自己的连接
    return conn.query(item_id)

with ThreadPoolExecutor(
    max_workers=10,
    initializer=init_connection,
    initargs=("postgresql://localhost/db",)
) as executor:
    results = list(executor.map(process_item, range(100)))

4. 限制 map 的内存使用

from concurrent.futures import ThreadPoolExecutor

def process(item):
    return item * 2

# 问题：map 的 timeout 针对每个结果，不是总时间
# 对于大输入，先分批处理
def chunked_map(executor, func, items, chunk_size=100):
    items = list(items)
    for i in range(0, len(items), chunk_size):
        chunk = items[i:i + chunk_size]
        yield from executor.map(func, chunk)

踩坑与注意事项

踩坑 1：multiprocessing 必须在 __main__ 保护下

在 Windows 和 spawn 启动方式下，子进程会重新导入主模块。如果在模块顶层直接创建进程，会导致无限递归创建子进程。

# 错误：不在 __main__ 保护下
from multiprocessing import Process

p = Process(target=some_func)
p.start()   # Windows 上会导致无限递归

# 正确
if __name__ == "__main__":
    p = Process(target=some_func)
    p.start()
    p.join()

踩坑 2：进程启动方式（spawn / fork / forkserver）

import multiprocessing

# fork（Linux/macOS 默认）：复制父进程，快但不安全（可能复制锁的状态）
# spawn（Windows 默认，macOS Python 3.8+ 默认）：全新子进程，慢但安全
# forkserver：通过专用服务器 fork，折中方案

# 显式设置启动方式
if __name__ == "__main__":
    multiprocessing.set_start_method("spawn")
    # 或者使用 context
    ctx = multiprocessing.get_context("spawn")
    p = ctx.Process(target=some_func)

踩坑 3：pickle 限制

进程间传递的数据必须可以被 pickle 序列化。以下对象不能被 pickle：

• lambda 函数（用 functools.partial 或模块级函数代替）
• 本地定义的函数（定义在函数内部的函数）
• 数据库连接、文件句柄、锁对象
• 某些第三方对象

from multiprocessing import Pool
from functools import partial

# 错误：lambda 无法被 pickle
with Pool() as pool:
    results = pool.map(lambda x: x * 2, range(10))  # PicklingError

# 正确：使用模块级函数或 partial
def double(x):
    return x * 2

with Pool() as pool:
    results = pool.map(double, range(10))

# 或者用 partial 传递额外参数
def multiply(x, factor):
    return x * factor

with Pool() as pool:
    results = pool.map(partial(multiply, factor=3), range(10))

踩坑 4：线程池中的异常会被静默吞掉

from concurrent.futures import ThreadPoolExecutor

def bad_task():
    raise RuntimeError("任务失败")

with ThreadPoolExecutor() as executor:
    future = executor.submit(bad_task)
    # 如果不调用 future.result()，异常会被静默忽略！

# 正确：总是检查 Future 的结果
with ThreadPoolExecutor() as executor:
    future = executor.submit(bad_task)

try:
    future.result()
except RuntimeError as e:
    print(f"捕获到异常: {e}")

踩坑 5：守护线程的陷阱

import threading
import time

def cleanup_worker():
    time.sleep(5)
    print("清理完成")  # daemon=True 时这行可能不会执行！

# daemon=True 的线程在主线程退出时被强制终止
t = threading.Thread(target=cleanup_worker, daemon=True)
t.start()
# 主线程立即退出，cleanup_worker 被强制终止

# 如果需要确保清理完成，不要用 daemon 或者在主线程 join
t = threading.Thread(target=cleanup_worker)
t.start()
t.join()  # 等待清理完成再退出

踩坑 6：fork 后不能使用 os.fork() 复制锁

import threading
import os

lock = threading.Lock()
lock.acquire()

# fork 会复制锁的状态，子进程中锁已被持有但没有线程持有它
# 这会导致死锁
pid = os.fork()
if pid == 0:
    # 子进程
    lock.acquire()  # 死锁！锁状态被复制，但持有锁的线程没有被复制

踩坑 7：Pool.map 会阻塞，大数据集需要注意内存

from multiprocessing import Pool

# 问题：map 收集所有结果到内存
with Pool() as pool:
    results = pool.map(process, range(10_000_000))  # 可能耗尽内存

# 更好：用 imap 或分批处理
with Pool() as pool:
    for result in pool.imap(process, range(10_000_000), chunksize=1000):
        handle(result)  # 逐个处理，不全部存入内存

常见陷阱

陷阱：threading.Thread 中未捕获的异常被静默吞噬

现象： 线程中的代码抛出异常，但主线程看不到任何报错，程序继续运行（状态错误）。
原因： 线程异常不会传播到主线程，Python 默认将其打印到 stderr 并终止该线程，不影响其他线程。
解决： 在线程函数中用 try/except 捕获，或使用 concurrent.futures.ThreadPoolExecutor 的 Future.result() 重新抛出异常：

with ThreadPoolExecutor() as pool:
    future = pool.submit(risky_task)
    result = future.result()  # 若任务抛出异常，此处会重新抛出

陷阱：multiprocessing 在 Windows 下 if __name__ == '__main__' 必须有

现象： Windows 上运行多进程程序时出现无限递归启动新进程（spawn 方式重新导入模块）。
原因： Windows 使用 spawn 启动子进程，会重新执行模块顶层代码，若没有 if __name__ == '__main__' 保护，主进程代码再次执行，无限创建子进程。
解决： 所有多进程启动代码必须放在 if __name__ == '__main__': 块内。

陷阱：asyncio 事件循环中执行同步阻塞调用

现象： 在 async 函数中调用同步 IO（requests.get、time.sleep）或 CPU 密集操作，导致整个事件循环卡住，其他协程无法运行。
原因： asyncio 单线程运行，同步阻塞调用会占用线程直到完成，期间所有协程暂停。
解决： 网络 IO 改用异步库（httpx、aiofiles），CPU 密集任务用 loop.run_in_executor 放到线程/进程池中运行。

参见

asyncio异步编程完全指南
contextlib完全指南