厨师互相等待：死锁是怎样炼成的

开篇除恐

提到“死锁”（Deadlock），很多人的脑海里会立刻浮现出两个线程互相持有对方需要的锁，谁也不肯放手，导致程序永远卡住的画面。听起来像是一种不可抗力的“厄运”，只要一发生就只能重启程序。其实，死锁并不是神秘的恶灵，它只是四个必要条件同时满足时的必然结果：互斥占有、占有并等待、不可抢占以及环路等待。把它想象成厨房里两位厨师，一人拿着锅铲等对方的调料罐，另一人拿着调料罐等对方的锅铲——谁也不肯先放手，结果两人都干不了活。只要了解这些条件以及如何打破它们，你就能在设计时就把死锁堵死在摇篮里，而不是事后去debug难以重现的卡死。

白话化

互斥占有（Mutual Exclusion）：资源（比如锁、厨具）同一时间只能被一个线程使用。
占有并等待（Hold and Wait）：线程已经持有至少一个资源，同时又在尝试获取其他已被其他线程占有的资源。
不可抢占（No Preemption）：已经占有的资源不能被强制夺走，只能由持有者自愿释放。
环路等待（Circular Wait）：存在一个线程链，链上的每个线程都在等待下一个线程所持有的资源，最后一个线程又在等待第一个线程的资源，形成一个环。

当这四个条件同时成立时，系统就会进入死锁状态。只要破坏其中任意一个条件，就能避免死锁。

直觉先行

想象两位厨师小明和小华在准备一道菜： - 小明需要先用锅铲，再用调料罐。 - 小华需要先用调料罐，再用锅铲。 - 厨房里只有一把锅铲和一个调料罐。

如果小明先拿到锅铲，小华先拿到调料罐，然后： - 小明发现调料罐被小华占有，于是等待调料罐； - 小华发现锅铲被小明占有，于是等待锅铲。

此时，两人都各自持有一件资源，并且在等待对方手中的资源——满足了占有并等待以及环路等待。由于锅铲和调料罐不能被抢走（不可抢占），而且每个资源同一时间只能被一人使用（互斥占有），四个条件全部具备，结果就是死锁：谁也没法继续做菜，只有外部干预（比如厨师长叫有人放手）才能打破。

如果我们事先约定一个资源使用顺序（比如所有厨师必须先拿锅铲再拿调料罐），那么即使小华也想先拿调料罐，他会被规则阻止，只能先等锅铲；只要锅铲空闲，他就会先拿锅铲，然后再去拿调料罐。这样就不可能出现环路等待，死锁自然就避免了。

例子贴身

例子 1（☼ 热身）：经典的两把锁死锁演示

任务：创建两个互斥锁 lock1 和 lock2，两条线程分别以不同顺序加锁，观察程序卡住。

怎么想到的：线程 A 先锁 lock1 再锁 lock2；线程 B 先锁 lock2 再锁 lock1。如果两线程几乎同时启动，很容易出现 A 持有 lock1 等 lock2，B 持有 lock2 等 lock1 的情况，导致互相等待，永远不会释放锁。

示例代码（Python）

import threading
import time

lock1 = threading.Lock()
lock2 = threading.Lock()

def worker_a():
    with lock1:                 # 获取 lock1
        time.sleep(0.01)        # 增加冲突概率
        with lock2:             # 尝试获取 lock2 -> 可能阻塞
            print("Worker A finished")

def worker_b():
    with lock2:                 # 获取 lock2
        time.sleep(0.01)
        with lock1:             # 尝试获取 lock1 -> 可能阻塞
            print("Worker B finished")

def main():
    t1 = threading.Thread(target=worker_a)
    t2 = threading.Thread(target=worker_b)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("Both finished")

if __name__ == "__main__":
    main()

运行结果：程序会卡住，永远不会打印 “Both finished”。因为每个线程都持有一把锁并在等另一把锁。

要点：这是最典型的死锁演示，只要加锁顺序不一致，就能 reproducible 地产生死锁。

例子 2（☼☼ 正经）：通过资源顺序避免死锁

任务：同上，但所有线程都同意按照相同的顺序（比如先 lock1 再 lock2）获取锁，确保不会死锁。

怎么想到的：我们把两把锁放进一个列表，并总是按锁的 id（或任何一致的标准）排序后再加锁。这样即使线程想要的锁是同两把，也不会出现循环等待。

示例代码（Python）

import threading
import time

def ordered_lock(*locks):
    """按锁的 id 排序后依次获取，避免死锁"""
    sorted_locks = sorted(locks, key=lambda x: id(x))
    for lock in sorted_locks:
        lock.acquire()
    try:
        yield
    finally:
        # 释放时按相反顺序释放（虽然不是必须，但保持对称）
        for lock in reversed(sorted_locks):
            lock.release()

def worker_safe(lock_a, lock_b, worker_id):
    # 使用上下文管理器确保即使出现异常也能释放锁
    with ordered_lock(lock_a, lock_b):
        time.sleep(0.005)
        print(f"Worker {worker_id} finished")

def main():
    lock1 = threading.Lock()
    lock2 = threading.Lock()
    threads = []
    for i in range(5):
        t = threading.Thread(target=worker_safe, args=(lock1, lock2, i))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("All finished safely")

if __name__ == "__main__":
    main()

运行结果：所有线程都能顺利完成，没有卡死。

要点：
- 强制全序（total order） 是避免死锁的最简单且有效的方法之一：只要所有线程申请资源的顺序完全一致，就不可能出现环路等待。
- 在实际项目中，常见的做法是为每种锁分配一个全局唯一的编号（比如按照 id 或哈希排序），然后在加锁前按照编号升序排序后再锁。
- 以上示例使用了自定义的 ordered_lock 上下文管理器来实现按 id 排序加锁。

例子 3（☼☼☼ 硬骨头）：带超时的尝试锁定（timeout）来破坏不可抢占条件

任务：使用 lock.acquire(timeout) 在获取锁失败时释放已持有的锁并重试，从而把“不可抢占”条件变为“可抢占”。

怎么想到的：线程在获取第一把锁后，尝试用 lock.acquire(timeout) 获取第二把锁；如果在超时时间内未能得到，则释放第一把锁，稍作休息后重新开始。这样即使出现了占有并等待的情况，也能够通过主动释放已持有的资源来打破死锁循环。

示例代码（Python）

import threading
import time

lock1 = threading.Lock()
lock2 = threading.Lock()

def worker_with_timeout():
    while True:
        # 按固定顺序获取第一把锁
        lock1.acquire()
        # 尝试在超时内获取第二把锁
        locked2 = lock2.acquire(timeout=0.01)  # 10 ms 超时
        if locked2:
            print("Worker got both locks")
            lock2.release()
            lock1.release()
            break
        # 第二把锁没得到，释放第一把锁后稍作等待重试
        lock1.release()
        time.sleep(0.005)  # 5 ms 休息后重试

def main():
    t1 = threading.Thread(target=worker_with_timeout)
    t2 = threading.Thread(target=worker_with_timeout)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("Both finished with timeout")

if __name__ == "__main__":
    main()

运行结果：程序能够顺利完成，尽管两线程一开始可能会争夺锁，但只要有一边在获取第二把锁时失败（超时），就会释放第一把锁，从而避免无限等待。

要点：
- lock.acquire(timeout) 提供了一种可抢占的方式：如果不能在规定时间内得到所需资源，就释放已有资源并稍后重试。
- 这种思路在实际中常用于数据库事务的死锁预防（事务回滚后重试）以及自适应互斥锁。

收尾

这一章要带走的东西
- 死锁的四个必要条件：互斥占有、占有并等待、不可抢占、环路等待。只有当这四个条件同时满足时，死锁才会发生。
- 最直接的预防办法是破坏环路等待：为所有资源规定一个全局顺序（如按 id 或其他一致的标准），所有线程必须按照此顺序申请资源。
- 其他策略包括：在申请新资源前先释放已有资源（破坏占有并等待）、使用带超时或 try_lock 的尝试锁定（将不可抢占变为可抢占）、采用死锁检测与恢复机制（在数据库等领域常见）。
- 在实际编码中，优先使用语言提供的高级同步原语（如 threading.Lock 配合上下文管理器，或自定义的有序加锁）来避免死锁。
- 掌握了死锁的成因与防备手段后，你就能在设计阶段就把潜在的死锁源头消灭，而不是事后去猜为什么程序忽然卡住。
- 下一章我们将探讨另一个经典的并发疑云——内存可见性问题：明明已经把值写进去了，为啥另一个线程还看不到？以及如何用内存屏障和原子操作保证正确的可见性。

就这样。 下一章我们来看看“厨房里的隐形猫”：内存可见性问题到底是怎么形成的，以及怎样让所有线程都能看到彼此的更新。