死锁必然发生，因代码满足互斥、持有并等待、不可抢占和循环等待四条件：线程1持lock_a等lock_b，线程2持lock_b等lock_a，形成循环依赖，导致双方永久阻塞。
死锁，在多线程编程里，它就像一个狡猾的陷阱，一旦触发，程序就会陷入无尽的等待。它不是一个“可能”发生的问题，而是在特定条件下“必然”会发生的一种僵局。本质上，就是两个或多个线程各自持有一把锁，同时又都在等待对方持有的另一把锁，形成一个循环依赖，谁也无法继续执行。
解决方案
import threading import time # 定义两把锁 lock_a = threading.Lock() lock_b = threading.Lock() def thread_function_1(): print("线程1: 尝试获取 lock_a...") lock_a.acquire() print("线程1: 已获取 lock_a，等待 0.1 秒...") time.sleep(0.1) # 引入短暂延迟，增加死锁发生的几率 print("线程1: 尝试获取 lock_b...") lock_b.acquire() print("线程1: 已获取 lock_b") # 执行一些操作 print("线程1: 正在执行任务...") lock_b.release() print("线程1: 已释放 lock_b") lock_a.release() print("线程1: 已释放 lock_a") print("线程1: 任务完成。") def thread_function_2(): print("线程2: 尝试获取 lock_b...") lock_b.acquire() print("线程2: 已获取 lock_b，等待 0.1 秒...") time.sleep(0.1) # 引入短暂延迟，增加死锁发生的几率 print("线程2: 尝试获取 lock_a...") lock_a.acquire() print("线程2: 已获取 lock_a") # 执行一些操作 print("线程2: 正在执行任务...") lock_a.release() print("线程2: 已释放 lock_a") lock_b.release() print("线程2: 已释放 lock_b") print("线程2: 任务完成。") if __name__ == "__main__": print("主线程: 启动线程1和线程2...") thread1 = threading.Thread(target=thread_function_1) thread2 = threading.Thread(target=thread_function_2) thread1.start() thread2.start() thread1.join() thread2.join() print("主线程: 所有线程已尝试完成。")
为什么这个特定的代码会导致死锁？
这个示例程序之所以必然导致死锁，是因为它完美地满足了死锁发生的四个经典条件：互斥、持有并等待、不可抢占和循环等待。让我们一步步剖析：
首先是互斥（Mutual Exclusion）：
threading.Lock
保证了同一时间只有一个线程能持有这把锁。
lock_a
和
lock_b
都具备这个特性，这是所有并发控制的基础。
接着是持有并等待（Hold and Wait）：当
thread_function_1
成功获取
lock_a
后，它并没有立即释放这把锁，而是继续尝试获取
lock_b
。同样，
thread_function_2
在获取
lock_b
后，也持有不放，转而等待
lock_a
。两个线程都持有一部分资源，同时又在等待另一部分资源，这就是“持有并等待”的体现。
然后是不可抢占（No Preemption）：我们使用的
threading.Lock
是不可抢占的。这意味着一旦一个线程获得了锁，除非它主动释放，否则其他线程无法强制夺走这把锁。操作系统或运行时环境不会介入去强制解除锁的持有状态。
最后，也是最关键的，是循环等待（Circular Wait）：
thread_function_1
正在等待
lock_b
，而
lock_b
正被
thread_function_2
持有。与此同时，
thread_function_2
正在等待
lock_a
，而
lock_a
正被
thread_function_1
持有。这形成了一个完美的环形依赖：线程1等线程2，线程2等线程1。它们都在等待对方释放自己需要的资源，但谁也无法释放自己持有的资源，因为它们都在等待。
time.sleep(0.1)
的引入，只是为了增加两个线程同时达到“持有并等待”状态的概率，让死锁更容易复现，但即使没有这个延迟，死锁在多核处理器上或不同的调度时机下也几乎是必然会发生的。
在多线程应用中，有哪些常见的模式可以有效避免死锁？
避免死锁，说起来容易做起来难，但核心原则是打破死锁的四个条件之一。在实际开发中，我们通常通过以下几种模式来规避：
一个非常有效且常用的策略是统一资源获取顺序。这意味着无论哪个线程，当它需要获取多个锁时，都应按照一个预先规定好的全局顺序来获取。比如，如果你的系统中有
lock_a
和
lock_b
，那么所有线程都应该先尝试获取
lock_a
，成功后再尝试获取
lock_b
。只要所有线程都遵循这个规则，就不会出现一个线程持有
lock_a
等
lock_b
，同时另一个线程持有
lock_b
等
lock_a
的情况，从而打破了循环等待条件。这个方法简单直接，在很多场景下都非常实用。
另一个思路是使用带超时机制的锁获取。比如
lock.acquire(timeout=some_value)
。如果线程在指定时间内未能获取到锁，它就会放弃等待，释放已经持有的资源，然后重新尝试，或者执行回退逻辑。这相当于打破了“持有并等待”的条件，因为它允许线程在无法获取所有必要资源时，主动释放已持有的资源，避免陷入无限期等待。当然，这需要更复杂的错误处理和重试机制。
更细粒度的锁控制有时也能帮助。不是所有操作都需要持有大范围的锁。通过识别代码中真正需要互斥保护的临界区，只在必要时才加锁，并尽快释放，可以减少锁的持有时间，从而降低死锁发生的概率。但要注意，过度细粒度的锁也可能导致性能下降和代码复杂度增加。
此外，避免在持有锁的情况下调用外部或未知代码。如果在一个持有锁的块中调用了可能需要其他锁的外部函数，或者该函数本身可能阻塞，那么死锁的风险就会大大增加。这要求我们对代码的依赖关系有清晰的认识。
最后，对于某些复杂的场景，可以考虑使用更高级的同步原语，比如
threading.RLock
（可重入锁），它允许同一个线程多次获取同一把锁而不会死锁自己。但这只解决了自死锁问题，并不能解决多个线程之间循环等待的死锁。还有一些并发库提供了更强大的工具，例如信号量、条件变量等，它们能以不同的方式管理资源的访问，有助于避免某些死锁场景。
如何有效地调试和识别正在运行程序中的死锁？
调试死锁是件令人头疼的事，因为它往往难以复现，且一旦发生，程序通常就“卡住”了，不报错也不退出。但也不是完全束手无策，以下是一些行之有效的方法：
首先，详细的日志记录是排查死锁的基石。在每次锁的获取 (
acquire
) 和释放 (
release
) 操作前后，都打印出详细的日志，包括线程ID、锁的名称、时间戳以及当前操作。当死锁发生时，通过分析这些日志，你可以清晰地看到哪些线程成功获取了哪些锁，又在等待哪些锁，从而勾勒出死锁的循环依赖路径。日志级别可以设置得高一些，比如
DEBUG
级别，在生产环境出现问题时可以动态开启。
其次，使用调试器。现代ide（如pycharm、VS Code）提供的多线程调试功能非常强大。你可以设置断点，暂停所有线程的执行，然后检查每个线程的调用栈和当前状态。在python中，如果一个线程被阻塞在
lock.acquire()
调用上，调试器通常能显示出它正在等待哪个锁。通过检查所有阻塞线程正在等待的锁，以及这些锁被哪个线程持有，你就能直接识别出死锁环。
对于已经运行的、卡死的Python程序，生成线程Dumps是一个非常实用的技巧。在linux/macOS上，你可以向Python进程发送
SIGQUIT
信号（
kill -QUIT <pid>
），Python解释器会将所有线程的堆栈信息打印到标准错误输出。在windows上，可以使用
Ctrl+break
（在命令行窗口）或者一些工具（如
pyrasite
或
WinDbg
配合Python扩展）。分析这些堆栈信息，寻找那些长时间停留在
acquire
或其他同步原语调用上的线程，它们很可能就是死锁的参与者。Python标准库的
faulthandler
模块也能在程序崩溃或收到特定信号时打印出所有线程的堆栈，这在调试生产环境问题时尤其有用。
最后，代码审查和静态分析也是预防死锁的重要手段。在代码提交前，进行严格的代码审查，特别是对涉及多线程和锁操作的代码块。遵循前面提到的“统一资源获取顺序”等模式，可以大大降低死锁的风险。虽然静态分析工具在检测死锁方面能力有限，但它们可以帮助发现一些潜在的并发问题。有时候，最有效的“调试”方式，就是从一开始就写出不会产生死锁的代码。
评论（已关闭）
评论已关闭