多线程

多线程是让一个进程同时管理多个执行路径。线程共享进程的地址空间，因此可以方便地访问同一批数据；操作系统会在多个线程之间切换，让等待网络、文件或数据库的线程不阻塞其他工作。

本文以 Python 3 和 CPython 为主要背景。CPython 执行普通 Python 字节码时受全局解释器锁（GIL）影响：同一进程内通常只有一个线程执行 Python 字节码。因此，多线程最适合 I/O 密集型任务；纯 Python 的 CPU 密集型任务通常应考虑多进程或专门释放 GIL 的计算库。

学完后，你应该能够：

创建、等待和停止线程；
使用锁、事件、条件和队列协调线程；
使用 ThreadPoolExecutor 管理一批相似任务；
正确处理异常、超时、共享状态和线程资源。

创建并等待线程

from threading import Thread
import time


def download(name: str, seconds: float) -> None:
    print(f"开始：{name}")
    time.sleep(seconds)  # 模拟 I/O 等待
    print(f"完成：{name}")


threads = [
    Thread(target=download, args=("文件 A", 0.3), name="download-a"),
    Thread(target=download, args=("文件 B", 0.2), name="download-b"),
]

for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

构造 Thread 只是在描述任务，调用 start() 才会创建并运行线程。join() 会阻塞当前线程，直到目标线程结束。不要直接调用 thread.run()，那只是普通方法调用，不会创建新线程。

常用属性和生命周期

name 便于日志定位，不能当作唯一标识；
ident 和 native_id 是运行期间的线程编号，可能被系统复用；
is_alive() 用于判断线程是否仍在运行；
daemon=True 表示主程序退出时不等待该线程，但不应依赖它完成数据写入或释放资源。

Python 没有安全地强制杀死任意线程的通用 API。需要停止时，应使用共享的 Event，让任务在安全位置主动退出：

from threading import Event, Thread
import time


def worker(stop: Event) -> None:
    while not stop.is_set():
        print("处理一批数据")
        if stop.wait(0.5):
            break
    print("线程收尾")


stop = Event()
thread = Thread(target=worker, args=(stop,))
thread.start()
time.sleep(1)
stop.set()
thread.join()

共享数据与竞态条件

线程共享普通对象，这很方便，也意味着多个线程可能同时修改同一个状态。即使某个操作看起来只有一行，也不能把“恰好一次完成”当成并发安全的保证：

from threading import Lock, Thread

balance = 0
lock = Lock()


def deposit() -> None:
    global balance
    for _ in range(10_000):
        with lock:
            balance += 1


threads = [Thread(target=deposit) for _ in range(4)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
print(balance)  # 40000

Lock 保护的是业务不变量，而不是某一行代码。with lock: 能保证正常退出和异常退出都会释放锁。需要同一线程重复进入时使用 RLock；但重复加锁通常也提示临界区设计过大。

其他同步原语

工具	用途
`Lock`	互斥访问共享资源
`RLock`	同一线程可以重复获取的锁
`Event`	一个线程向其他线程广播“已发生/请停止”
`Condition`	等待某个状态满足，再被通知唤醒
`Semaphore`	限制同时进入资源的线程数
`Barrier`	等待固定数量的线程到达同一阶段
`queue.Queue`	线程安全地传递任务或结果

用 `Condition` 等待状态变化

条件变量把“保护状态的锁”和“状态变化通知”组合起来。等待方应使用 wait_for() 反复检查条件，而不是假设被唤醒就一定满足条件：

from threading import Condition, Thread

condition = Condition()
items: list[str] = []


def consumer() -> None:
    with condition:
        condition.wait_for(lambda: bool(items))
        print("取得：", items.pop())


thread = Thread(target=consumer)
thread.start()
with condition:
    items.append("任务")
    condition.notify()
thread.join()

notify() 唤醒一个等待者，notify_all() 唤醒全部等待者；它们不会释放锁，被唤醒的线程要等通知方退出 with 后才能继续。

用 `Semaphore` 限制并发量

信号量允许固定数量的线程同时进入，适合保护数据库连接数或外部服务并发额度：

from threading import BoundedSemaphore

slots = BoundedSemaphore(3)


def call_service() -> None:
    with slots:
        send_request()

BoundedSemaphore 会在释放次数超过获取次数时抛出错误，通常比普通 Semaphore 更容易发现配对错误。它只限制同时执行数量，不能替代超时、重试或服务端限流。

优先使用消息队列减少共享可变状态。典型的生产者—消费者结构如下：

from queue import Queue
from threading import Thread

tasks: Queue[int | None] = Queue()


def consumer() -> None:
    while True:
        item = tasks.get()
        try:
            if item is None:
                return
            print(item * item)
        finally:
            tasks.task_done()


thread = Thread(target=consumer)
thread.start()
for number in range(5):
    tasks.put(number)
tasks.put(None)
tasks.join()
thread.join()

task_done() 必须与每次 get() 配对，否则 join() 会一直等待。队列中的结束标记（这里是 None）应按消费者数量放入足够多份。

Queue(maxsize=n) 还能提供背压：队列已满时，生产者的 put() 会等待，避免生产速度长期超过消费速度而耗尽内存。

线程池：批量执行任务

大量同类任务不应为每一项反复创建线程。concurrent.futures.ThreadPoolExecutor 会复用固定数量的工作线程：

from concurrent.futures import ThreadPoolExecutor, as_completed


def fetch(url: str) -> str:
    return f"已获取 {url}"


urls = ["https://example.com/a", "https://example.com/b"]
with ThreadPoolExecutor(max_workers=8, thread_name_prefix="fetch") as executor:
    futures = [executor.submit(fetch, url) for url in urls]
    for future in as_completed(futures):
        try:
            print(future.result())
        except Exception as error:
            print("任务失败：", error)

executor.map() 保持输入顺序，submit() 配合 as_completed() 则可以先处理先完成的任务。调用 future.result() 会重新抛出工作线程中的异常，因此不要只检查 Future 是否创建成功。

超时与取消

from concurrent.futures import TimeoutError

future = executor.submit(fetch, "https://example.com/slow")
try:
    value = future.result(timeout=2)
except TimeoutError:
    cancelled = future.cancel()

result(timeout=...) 超时不会自动停止任务。cancel() 只能取消尚未开始的任务，不能强制终止已经运行的线程。因此任务内部使用的网络、锁和队列操作也必须设置超时，并定期检查停止信号。

使用 executor.shutdown(wait=True, cancel_futures=True) 可以取消尚未开始的 Future 并等待正在运行的任务结束；with 上下文默认会等待已提交任务完成。

线程池大小取决于等待时间、外部服务限流和本机资源。线程越多不一定越快；过多线程会增加上下文切换、内存和对方服务的压力。对网络请求还应设置连接、读取和整体超时。

异常、日志与调试

手动创建的线程如果抛出未捕获异常，默认会调用 threading.excepthook 并打印堆栈，但异常不会自动在创建线程的代码中重新抛出。需要可靠收集结果时，优先使用 Future；也可以在任务内部捕获异常并放入结果队列。

日志中加入 %(threadName)s 能帮助区分并发任务。测试并发代码时，应使用 Event、Barrier 或队列建立确定的执行顺序，不要依赖 sleep() 猜测线程已经运行到哪一步。

死锁发生时，可以用 faulthandler.dump_traceback_later() 定时打印所有线程的调用栈，或在调试器中检查每个线程正在等待的锁。关键是记录“持有什么锁、等待什么资源”，而不只是记录函数入口。

GIL 的准确理解

GIL 是 CPython 解释器保护内部状态的一把锁，并不等于 Python 对象天然线程安全，也不等于所有线程都不能并行：

执行普通 Python 字节码时，同一解释器中通常只有一个线程持有 GIL；
阻塞 I/O 和许多 C 扩展会释放 GIL，让其他线程执行；
某个内置操作在当前 CPython 版本中看似原子，不代表多个步骤组成的业务逻辑也是原子的；
不同 Python 实现和未来运行模式可能具有不同约束，正确性不应依赖偶然的字节码实现细节。

因此，共享数据是否需要同步，应由业务操作是否必须保持一致来决定，而不是根据 GIL 猜测。

常见问题与实践建议

不要让线程无期限等待锁、队列或网络；使用 timeout 并记录上下文。
不要在持有锁时执行慢 I/O，否则其他线程都会被拖住。
避免锁顺序不一致导致死锁；需要多个锁时统一获取顺序。
线程之间共享的容器要么由锁保护，要么改为通过 Queue 传递消息。
在线程池中不要让任务无限等待另一个同一个线程池中的任务，线程数耗尽后会互相等待。
线程适合并发等待，不等于可以让纯 Python 计算在一个进程中真正并行。

如果任务主要是 CPU 计算，应比较 ProcessPoolExecutor；如果任务可以拆成非阻塞 I/O，应比较 asyncio。选择并发模型时，以实际测量的等待、计算、通信和内存成本为依据。

创建并等待线程​

常用属性和生命周期​

共享数据与竞态条件​

其他同步原语​

用 Condition 等待状态变化​

用 Semaphore 限制并发量​

线程池：批量执行任务​

超时与取消​

异常、日志与调试​

GIL 的准确理解​

常见问题与实践建议​