抢锅铲的灾难：竞态条件

开篇除恐

听到“竞态条件”（Race Condition）这四个字，很多人会立刻联想到无法调试的诡异bug、只在凌晨三点出现一次的崩溃，或者多次运行结果都不一样的程序。听起来像是某种难以捉摸的幽灵，让人对多线程编程望而生畏。其实，竞态条件并不是魔法，它只是“多个线程在没有同步的情况下，同时去读写同一份数据，导致结果依赖于执行的精确时序”。把它想象成厨房里两个厨师都抓住同把锅铲去翻菜——谁先抢到、谁后拿、以及他们各自动作的细微交错，都会决定最终菜是翻匀了还是撒了一地。只要明白“谁先动手”和“动作是不是原子的”这两点，就能够用锁或者恰当的调度顺序来避免灾难。

白话话

竞态条件的核心：两个或以上的线程对同一内存位置进行读取-修改-写入（Read‑Modify‑Write，简称 RMW）操作，而这些步骤不是原子的（即不能在一个不可分割的时刻完成）。如果在这段时间里另一个线程也读取了旧值，则后来的写入会把先前的更新覆盖掉，导致丢失更新。
典型例子：counter += 1 在 Python 中实际上是：读取 counter 的值 → 在寄存器里加一 → 把结果写回内存。如果两个线程几乎同时执行这三步，可能得到的最终值只比原值大一而不是大二。
避免手段：
1. 使用互斥锁把临界区保护起来：只有拿到锁的线程才能执行读取‑修改‑写入，其他线程必须等待。
2. 使用队列（Queue）或其他线程安全的数据结构：在生产者‑消费者等场景中，队列内部已经实现了必要的同步。
3. 在特定场景下使用原子操作的库：虽然 Python 没有内置的原子整数，但可以使用 multiprocessing.Value 或第三方库（如 atomics）来实现；然而最常见且简单的做法仍是使用 threading.Lock。

只要记住：竞态不是因为线程“太快”，而是因为对共享数据的访问没有被正确地串行化。

直觉先行

想象你和朋友共用一个计票器：每人手里有一张票，要把票数加到同一个计数器上。计数器的工作步骤是： 1. 看一下当前显示的数字（读取）； 2. 在心里把这个数字加一（修改）； 3. 把新的数字写回显示屏（写入）。

如果你们几乎同时进行这三步，可能会出现这样的情形： - 你读到 5，朋友也读到 5（因为还没来得及写回）； - 你心里算得 6，朋友也算得 6； - 你把 6 写回显示屏，朋友也把 6 写回显示屏； - 最终显示屏上是 6，而不是应有的 7。这就是一次“丢失更新”的竞态条件。

如果给计数器装上一个锁：只有拿到锁的人才能进行读取‑修改‑写入的完整三步，其他人必须在旁边等着。那么不管你们多快，最终的结果总是准确的。

例子贴身

例子 1（☼ 热身）：未保护的递增导致丢失更新

任务：启动 2 条线程，各自对 counter 加 5000 次，观察最终结果经常小于 10000。

怎么想到的：我们故意不加任何同步，只让每条线程在循环里执行 counter += 1。然后多次运行程序，统计结果的分布。

示例代码（Python）

import threading
import time

counter = 0  # 非原子，非锁保护

def worker():
    global counter
    for _ in range(5000):
        counter += 1  # 可能产生竞态

def main():
    threads = []
    for _ in range(2):
        t = threading.Thread(target=worker)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print(f"Final counter = {counter}")

if __name__ == "__main__":
    main()

多次运行结果（示例）

Final counter = 9832
Final counter = 9915
Final counter = 9978
...

几乎每次都小于 10000，说明有更新被丢失。

例子 2（☼☼ 正经）：用互斥锁保证正确递增

任务：同上，但使用互斥锁确保每次递增都是受保护的，最终必然等于 10000。

怎么想到的：我们给 counter 配一个互斥锁 lock；每次要改动 counter 时，先把锁锁上，改完再解锁。这样保证同一时刻只有一条线程能访问 counter。

示例代码（Python）

import threading

counter = 0
lock = threading.Lock()

def worker():
    global counter
    for _ in range(5000):
        with lock:          # 自动加锁/解锁（RAII 风格）
            counter += 1

def main():
    threads = []
    for _ in range(2):
        t = threading.Thread(target=worker)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print(f"Final counter = {counter}")  # 期望 10000

if __name__ == "__main__":
    main()

运行结果

Final counter = 10000

要点：
- 互斥锁是防止竞态条件的最直接手段；with lock: 确保即使在异常情况下也会释放锁。
- 虽然 Python 没有内置的原子整数，但通过锁可以实现同样的效果；在需要高性能且只做简单递增的场景，也可以考虑使用 queue.Queue 或 multiprocessing.Value 等线程安全的结构。

例子 3（☼☼☼ 硬骨头）：自旋锁的简单实现（概念演示）

任务：展示如何用一个布尔标志和忙等待实现一个自旋锁，来说明即使没有现成的互斥锁，也能通过底层同步原理达成互斥。（此例仅为概念展示，实际生产中请使用 threading.Lock。）

怎么想到的：自旋锁的核心是一个布尔标志 locked（False 表示未锁定，True 表示已锁定）。线程想要进入临界区时，不断检查该标志：如果为 False，则将其设为 True 并进入临界区；如果已经是 True，则表示有人持有锁，线程继续循环（自旋），直到拿到锁。退出时把标志位恢复为 False。为了避免忙等待消耗过多 CPU，可在循环中加入极短的 time.sleep(0) 让出 GIL。

示例代码（Python）

import threading
import time

class SpinLock:
    def __init__(self):
        self._locked = False
        self._mutex = threading.Lock()  # 用来保护 _locked 的原子更新（实际中可用 threading.Lock 本身，此处仅演示概念）

    def acquire(self):
        while True:
            with self._mutex:
                if not self._locked:
                    self._locked = True
                    return
            # 自旋：极短暂停让出 GIL，避免占满 CPU
            time.sleep(0.0001)

    def release(self):
        with self._mutex:
            self._locked = False

def worker(lock_obj, shared_counter, iterations):
    for _ in range(iterations):
        lock_obj.acquire()
        try:
            shared_counter[0] += 1   # 临界区
        finally:
            lock_obj.release()

def main():
    shared = [0]  # 用列表使其可在线程间修改
    lock = SpinLock()
    threads = []
    iters = 2500
    for _ in range(4):
        t = threading.Thread(target=worker, args=(lock, shared, iters))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print(f"Shared final = {shared[0]}")  # 期望 10000

if __name__ == "__main__":
    main()

要点：
- 自旋锁在竞争不激烈且持锁时间极短时可以避免线程挂起和唤醒的开销；但若持锁时间较长，会浪费 CPU 循环。
- 这说明即使没有现成的 threading.Lock，也可以通过底层的同步原理构建互斥机制——这正是操作系统内部实现互斥锁的基本思路。实际编程中，直接使用 threading.Lock 更简单、安全且高效。

收尾

这一章要带走的东西
- 竞态条件源于多线程对同一数据的非原子读‑修改‑写入操作的交错，导致结果依赖于不确定的执行时序。
- 常见的表现包括：丢失更新、脏读、不一致的状态。
- 防止竞态的主要手段：使用互斥锁保护临界区、使用线程安全的数据结构（如 Queue），在特殊场景下可考虑自旋锁或其他原子操作的实现。
- 理解了竞态条件的本质后，你就能够有针对性地检查代码：凡是出现读取‑修改‑写入的模式（如 counter += 1, counter = counter + 1, flag |= bit，读取后再根据旧值做决定），都要问自己：“这里是否需要同步？”
- 下一章我们将探讨另一个让人头疼的并发故障——死锁：为什么两个线程各自持有资源并等待对方会导致永久卡住，以及如何通过资源顺序、超时等策略来避免。

就这样。 下一章我们来看看“厨师互相等待”：死锁是怎样炼成的，以及怎样在设计时就把它堵死在摇篮里。