Python多线程实战：threading模块详解、线程安全与锁机制、死锁的预防与排查

聊到Python并发，很多人第一反应是「GIL限制，多线程没用」。这话对了一半。GIL确实让CPU密集型任务的多线程收益有限，但I/O密集型场景——比如网络请求、文件读写、数据库操作——多线程依然是利器。我这些年做高并发服务，Python多线程用得最多的恰恰是这类场景。

这一章，咱们把Python的threading模块掰开揉碎。从基础用法到锁机制，再到死锁的排查与预防，我会把项目中踩过的坑一并分享出来。

2.1 threading模块核心组件

Python的threading模块提供了多线程编程的基础设施。我个人习惯把它分成三类：线程对象、同步原语、工具函数。

组件	用途	我常用的场景
`Thread`	创建和管理线程	启动后台任务、并行I/O
`Lock`	互斥锁，保护共享资源	计数器、缓存更新
`RLock`	可重入锁，同一线程可多次获取	递归函数中的资源保护
`Semaphore`	信号量，控制并发数量	连接池、限流
`Event`	线程间事件通知	优雅关闭、任务协调
`Condition`	条件变量，等待/通知模式	生产者-消费者模型

先看一个最基础的线程创建示例：

import threading
import time

def worker(name, delay):
    print(f"线程 {name} 启动")
    time.sleep(delay)
    print(f"线程 {name} 结束")

# 创建线程
t1 = threading.Thread(target=worker, args=("A", 2))
t2 = threading.Thread(target=worker, args=("B", 1))

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

# 等待线程结束
t1.join()
t2.join()

print("所有线程执行完毕")

这里有个细节：join()是必须的吗？不一定。但如果你需要等待子线程完成后再继续主线程逻辑，就得调用它。我在项目中经常用join(timeout)来设置超时，防止线程卡死导致主线程无限等待。

小技巧： 使用 threading.enumerate() 可以列出当前所有存活的线程，调试时非常有用。我曾经用它快速定位到某个忘记关闭的后台线程。

2.2 线程安全与锁机制

多线程最头疼的问题是什么？数据竞争。两个线程同时修改同一个变量，结果可能不是你想要的。看个经典例子：

counter = 0

def increment():
    global counter
    for _ in range(100000):
        counter += 1

threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"期望值: 1000000, 实际值: {counter}")
# 输出可能只有 800000 左右

为什么会这样？counter += 1 在Python字节码层面是三条指令：读取、加1、写入。线程切换可能发生在任意两条指令之间，导致更新丢失。

解决方案就是加锁：

lock = threading.Lock()
counter = 0

def safe_increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

# 结果稳定在 1000000

这里我强烈推荐使用with lock:上下文管理器。它自动处理获取和释放，即使中间抛出异常也不会死锁。我见过太多新手手动调用acquire()和release()，结果漏掉release()导致程序卡死。

2.2.1 Lock vs RLock

Lock是普通互斥锁，同一线程不能重复获取。如果你在递归函数里用Lock，第二次acquire()会阻塞自己——死锁了。

RLock（可重入锁）允许同一线程多次获取，内部维护一个计数器。每次acquire()加1，release()减1，减到0才真正释放。

def recursive_lock_example(lock, depth):
    with lock:
        if depth > 0:
            recursive_lock_example(lock, depth - 1)

# Lock 会死锁，RLock 正常工作
rlock = threading.RLock()
recursive_lock_example(rlock, 5)  # 没问题

注意： 能用 Lock 就别用 RLock。RLock 会掩盖设计问题，比如锁的粒度太粗。我一般只在递归或回调场景下才用 RLock。

2.3 死锁的预防与排查

死锁是并发编程的噩梦。四个必要条件：互斥、持有并等待、不可剥夺、循环等待。破坏任意一个就能预防死锁。

我在项目中遇到过最典型的死锁场景：两个线程各自持有一个锁，然后等待对方释放。看代码：

lock_a = threading.Lock()
lock_b = threading.Lock()

def thread_1():
    with lock_a:
        time.sleep(0.1)
        with lock_b:
            print("线程1完成")

def thread_2():
    with lock_b:
        time.sleep(0.1)
        with lock_a:
            print("线程2完成")

# 大概率死锁

预防死锁，我总结了三条实战经验：

固定锁顺序：所有线程按相同顺序获取锁。比如都先拿lock_a再拿lock_b。
使用超时：lock.acquire(timeout=5)，获取不到就放弃，避免无限等待。
锁的粒度尽量小：能锁变量就别锁对象，能锁对象就别锁整个函数。

    避坑指南： 我曾经排查过一个线上死锁问题，耗时整整半天。最后发现是第三方库内部用了 Lock，而我们的回调函数里又去获取了同一个锁。解决方案是把回调放到线程池里异步执行，打破循环等待。

2.3.1 死锁排查工具

Python提供了threading模块的调试能力。当程序卡住时，可以发送SIGQUIT信号（Linux下kill -3 PID）让Python打印所有线程的堆栈信息。或者用faulthandler模块：

import faulthandler
import signal

faulthandler.register(signal.SIGUSR1)

# 运行时发送 kill -USR1 PID 即可打印所有线程堆栈

另一个实用技巧：用threading._active字典查看当前所有活跃线程。虽然以下划线开头，但调试时用用无妨。

2.4 本章知识体系

下面这张图总结了多线程编程的核心脉络，从基础组件到锁机制再到死锁预防，你可以对照着梳理自己的知识结构：

2.5 实战建议

最后，分享几个我在项目中沉淀下来的多线程编码习惯：

能用线程池就别手动创建线程：concurrent.futures.ThreadPoolExecutor 管理起来更省心。
共享数据尽量用不可变对象：比如用元组代替列表，用namedtuple代替字典。
锁的持有时间越短越好：只锁关键操作，不要在锁里做I/O或复杂计算。
写单元测试时故意制造并发竞争：用threading.Barrier让多个线程同时到达临界点，更容易暴露问题。

一个实用工具： Python 3.8+ 的 threading.Thread 支持 name 参数。给每个线程起个有意义的名字，排查问题时一眼就能看出哪个线程在干什么。

多线程编程，说白了就是管理「共享」和「同步」的艺术。锁不是越多越好，也不是越少越好。找到合适的粒度，保持代码可读性，比追求极致性能更重要。毕竟，一个跑得慢但正确的程序，远好过一个跑得快但偶尔崩溃的程序。

📚 课程目录

Python多线程实战：threading模块详解、线程安全与锁机制、死锁的预防与排查

2.1 threading模块核心组件

2.2 线程安全与锁机制

2.2.1 Lock vs RLock

2.3 死锁的预防与排查

2.3.1 死锁排查工具

2.4 本章知识体系

2.5 实战建议