> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# 并发编程完全指南
- URL: https://blog.vercanti.com/bing-fa-bian-cheng-wan-quan-zhi-nan/
- Published: 2026-08-28T14:34:35.000Z
- Updated: 2026-08-28T14:56:53.000Z
- Description: Python 提供三种主要并发模型：基于线程的 threading、基于进程的 multiprocessing 和基于协程的 asyncio。此外，concurrent.futures 提供了统一的高层接口。选择哪种模型取决于任务类型。 CPython 的 GIL 确保同一时刻只有一个线程执行 Python 字节码。因此，对于纯 Python 的 CPU 密集型代码，多线程不能带来并行加速。但对于 IO 密集型任务，线程在等待 IO 时会释放 GIL，其他线程可以运行。 C 扩展（如 NumPy 的计算部分）在执行期间可以主动释放 GIL，实现真正的并行
- Author: yellowdog
- Tags: Python, 基础

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

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

## 并发模型选择指南

| 场景                | 推荐方案                                  | 原因                  |
| ----------------- | ------------------------------------- | ------------------- |
| IO 密集型（网络请求、文件读写） | ThreadPoolExecutor 或 asyncio          | 等待 IO 时释放 GIL 或切换协程 |
| CPU 密集型（计算、图像处理）  | ProcessPoolExecutor 或 multiprocessing | 绕过 GIL，真正并行运行       |
| 大量并发 IO，需要低开销     | asyncio                               | 协程比线程开销小，单线程处理数千并发  |
| 简单后台任务            | threading.Thread                      | 轻量，共享内存，适合简单场景      |
| 批量数据处理            | multiprocessing.Pool                  | map/starmap 接口简洁    |

---

## threading 模块

### GIL（全局解释器锁）

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

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

### Thread 基础

```python
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` 的构造参数：

| 参数     | 类型              | 默认值  | 说明                  |
| ------ | --------------- | ---- | ------------------- |
| group  | None            | None | 保留参数，始终为 None       |
| target | callable 或 None | None | 线程运行的函数             |
| name   | str 或 None      | None | 线程名，默认自动生成 Thread-N |
| args   | tuple           | ()   | 传给 target 的位置参数     |
| kwargs | dict            | {}   | 传给 target 的关键字参数    |
| daemon | bool 或 None     | None | 是否为守护线程；None 则继承父线程 |

`join` 方法的参数：

| 参数      | 类型           | 默认值  | 说明                 |
| ------- | ------------ | ---- | ------------------ |
| timeout | float 或 None | None | 等待超时秒数，None 表示无限等待 |

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

### 继承 Thread 类

```python
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` 保证同一时刻只有一个线程访问共享资源，防止竞态条件。

```python
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` 的参数：

| 参数       | 类型    | 默认值  | 说明                 |
| -------- | ----- | ---- | ------------------ |
| blocking | bool  | True | False 时立即返回（非阻塞模式） |
| timeout  | float | \-1  | 等待超时秒数，\-1 表示无限等待  |

### RLock（可重入锁）

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

```python
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` 限制同时访问资源的线程数量，适用于限制并发连接数等场景。

```python
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` 的构造参数：

| 参数    | 类型  | 默认值 | 说明                    |
| ----- | --- | --- | --------------------- |
| value | int | 1   | 内部计数器的初始值，即允许同时进入的线程数 |

### Event（事件）

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

```python
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` 的主要方法：

| 方法                 | 说明                     |
| ------------------ | ---------------------- |
| set()              | 将内部标志设为 True，唤醒所有等待的线程 |
| clear()            | 将内部标志重置为 False         |
| is\_set()          | 返回内部标志的布尔值             |
| wait(timeout=None) | 阻塞直到标志为 True 或超时，返回标志值 |

`wait` 的参数：

| 参数      | 类型           | 默认值  | 说明               |
| ------- | ------------ | ---- | ---------------- |
| timeout | float 或 None | None | 超时秒数，None 表示无限等待 |

### Condition（条件变量）

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

```python
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` 的参数：

| 参数      | 类型           | 默认值  | 说明   |
| ------- | ------------ | ---- | ---- |
| timeout | float 或 None | None | 超时秒数 |

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

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

```python
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 基础

```python
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` 的构造参数：

| 参数     | 类型              | 默认值  | 说明               |
| ------ | --------------- | ---- | ---------------- |
| group  | None            | None | 保留参数，始终为 None    |
| target | callable 或 None | None | 子进程运行的函数         |
| name   | str 或 None      | None | 进程名称             |
| args   | tuple           | ()   | 传给 target 的位置参数  |
| kwargs | dict            | {}   | 传给 target 的关键字参数 |
| daemon | bool 或 None     | None | 是否为守护进程          |

`join` 方法的参数：

| 参数      | 类型           | 默认值  | 说明     |
| ------- | ------------ | ---- | ------ |
| timeout | float 或 None | None | 等待超时秒数 |

### Pool（进程池）

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

```python
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` 的构造参数：

| 参数               | 类型               | 默认值           | 说明                     |
| ---------------- | ---------------- | ------------- | ---------------------- |
| processes        | int 或 None       | None（CPU 核心数） | 工作进程数量                 |
| initializer      | callable 或 None  | None          | 每个工作进程启动时调用的初始化函数      |
| initargs         | tuple            | ()            | 传给 initializer 的参数     |
| maxtasksperchild | int 或 None       | None          | 每个工作进程处理多少任务后重启，防止内存泄漏 |
| context          | context 对象或 None | None          | 启动方式上下文                |

`map` 方法的参数：

| 参数        | 类型         | 默认值  | 说明                         |
| --------- | ---------- | ---- | -------------------------- |
| func      | callable   | 必填   | 要应用的函数                     |
| iterable  | iterable   | 必填   | 输入数据                       |
| chunksize | int 或 None | None | 每次发送给工作进程的任务数，较大值减少进程间通信开销 |

`starmap` 方法的参数：

| 参数        | 类型                 | 默认值  | 说明                       |
| --------- | ------------------ | ---- | ------------------------ |
| func      | callable           | 必填   | 要应用的函数                   |
| iterable  | iterable of tuples | 必填   | 每个元素是一个元组，解包后作为参数传给 func |
| chunksize | int 或 None         | None | 每次发送的任务块大小               |

### Queue 和 Pipe（进程间通信）

```python
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` 的构造参数：

| 参数      | 类型  | 默认值 | 说明            |
| ------- | --- | --- | ------------- |
| maxsize | int | 0   | 队列最大容量，0 表示无限 |

`Pipe` 的参数：

| 参数     | 类型   | 默认值  | 说明                                      |
| ------ | ---- | ---- | --------------------------------------- |
| duplex | bool | True | True 为双向管道，False 为单向（conn1 只读，conn2 只写） |

### Manager（共享对象）

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

```python
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 类型数据。

```python
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` 的构造参数：

| 参数                 | 类型              | 默认值  | 说明                                        |
| ------------------ | --------------- | ---- | ----------------------------------------- |
| typecode\_or\_type | str 或 ctypes 类型 | 必填   | 数据类型，如 'i'（int）、'd'（double）、ctypes.c\_int |
| \*args             | any             | \-   | 传给类型构造函数的初始值                              |
| lock               | bool 或 Lock     | True | 是否自动添加锁，True 创建新锁，也可传入现有锁                 |

`Array` 的构造参数：

| 参数                    | 类型              | 默认值  | 说明         |
| --------------------- | --------------- | ---- | ---------- |
| typecode\_or\_type    | str 或 ctypes 类型 | 必填   | 元素类型       |
| size\_or\_initializer | int 或 iterable  | 必填   | 数组大小或初始值序列 |
| lock                  | bool 或 Lock     | True | 是否自动添加锁    |

---

## concurrent.futures 模块

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

### ThreadPoolExecutor

```python
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` 的构造参数：

| 参数                   | 类型              | 默认值                   | 说明                 |
| -------------------- | --------------- | --------------------- | ------------------ |
| max\_workers         | int 或 None      | None（min(32, CPU数+4)） | 最大线程数              |
| thread\_name\_prefix | str             | ""                    | 线程名称前缀，便于调试        |
| initializer          | callable 或 None | None                  | 每个线程启动时调用的初始化函数    |
| initargs             | tuple           | ()                    | 传给 initializer 的参数 |

### ProcessPoolExecutor

```python
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` 的构造参数：

| 参数                     | 类型              | 默认值           | 说明                              |
| ---------------------- | --------------- | ------------- | ------------------------------- |
| max\_workers           | int 或 None      | None（CPU 核心数） | 最大进程数                           |
| mp\_context            | context 或 None  | None          | 多进程启动上下文（spawn/fork/forkserver） |
| initializer            | callable 或 None | None          | 每个工作进程启动时调用的初始化函数               |
| initargs               | tuple           | ()            | 传给 initializer 的参数              |
| max\_tasks\_per\_child | int 或 None      | None          | 每个子进程最多执行的任务数（Python 3.11+）     |

### submit 方法

| 参数         | 类型       | 默认值 | 说明           |
| ---------- | -------- | --- | ------------ |
| fn         | callable | 必填  | 要执行的函数       |
| \*args     | any      | \-  | 传给 fn 的位置参数  |
| \*\*kwargs | any      | \-  | 传给 fn 的关键字参数 |

返回 `Future` 对象。

### map 方法

| 参数          | 类型           | 默认值  | 说明                                 |
| ----------- | ------------ | ---- | ---------------------------------- |
| fn          | callable     | 必填   | 要应用的函数                             |
| \*iterables | iterable     | 必填   | 输入数据，可以是多个迭代器（对应函数的多个参数）           |
| timeout     | float 或 None | None | 每个结果的获取超时时间（秒）                     |
| chunksize   | int          | 1    | 仅 ProcessPoolExecutor 有效，批量提交任务的大小 |

### as\_completed 函数

| 参数      | 类型                 | 默认值  | 说明                      |
| ------- | ------------------ | ---- | ----------------------- |
| fs      | iterable of Future | 必填   | Future 对象的集合            |
| timeout | float 或 None       | None | 等待超时秒数，超时抛 TimeoutError |

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

### wait 函数

```python
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` 函数的参数：

| 参数           | 类型                 | 默认值            | 说明                                                    |
| ------------ | ------------------ | -------------- | ----------------------------------------------------- |
| fs           | iterable of Future | 必填             | Future 对象集合                                           |
| timeout      | float 或 None       | None           | 等待超时秒数                                                |
| return\_when | str 常量             | ALL\_COMPLETED | 返回时机：ALL\_COMPLETED、FIRST\_COMPLETED、FIRST\_EXCEPTION |

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

### Future 对象的方法

```python
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` 方法的参数：

| 参数      | 类型           | 默认值  | 说明                                         |
| ------- | ------------ | ---- | ------------------------------------------ |
| timeout | float 或 None | None | 等待超时秒数，超时抛 concurrent.futures.TimeoutError |

`exception` 方法的参数：

| 参数      | 类型           | 默认值  | 说明     |
| ------- | ------------ | ---- | ------ |
| timeout | float 或 None | None | 等待超时秒数 |

`add_done_callback` 方法的参数：

| 参数 | 类型       | 默认值 | 说明                  |
| -- | -------- | --- | ------------------- |
| fn | callable | 必填  | 回调函数，接受一个 Future 参数 |

### ProcessPoolExecutor vs multiprocessing.Pool 对比

| 特性              | ProcessPoolExecutor          | multiprocessing.Pool |
| --------------- | ---------------------------- | -------------------- |
| API 风格          | 高层，统一接口                      | 低层，功能丰富              |
| 异常处理            | future.result() 重新抛出         | 需手动检查                |
| 取消任务            | future.cancel()              | 不支持                  |
| imap\_unordered | 不支持                          | 支持                   |
| starmap         | 不支持（用 lambda 或 partial）      | 支持                   |
| 与 asyncio 集成    | 原生支持（loop.run\_in\_executor） | 需要额外包装               |
| 初始化器            | 支持                           | 支持                   |

---

## 最佳实践

### 1\. 使用 context manager 管理 Executor

```python
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\. 正确处理异常

```python
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 共享昂贵资源

```python
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` 的内存使用

```python
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` 启动方式下，子进程会重新导入主模块。如果在模块顶层直接创建进程，会导致无限递归创建子进程。

```python
# 错误：不在 __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）

```python
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` 或模块级函数代替）
- 本地定义的函数（定义在函数内部的函数）
- 数据库连接、文件句柄、锁对象
- 某些第三方对象

```python
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：线程池中的异常会被静默吞掉

```python
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：守护线程的陷阱

```python
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()` 复制锁

```python
import threading
import os

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

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

```

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

```python
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()` 重新抛出异常：

```python
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异步编程完全指南](https://blog.vercanti.com/asyncio-yi-bu-bian-cheng-wan-quan-zhi-nan/)  
[contextlib完全指南](https://blog.vercanti.com/contextlib-wan-quan-zhi-nan/)