为何我编写的非线程安全Python代码表现出线程安全特性？

问题背景 #

编写了一个未做线程同步的ConcurrentQueue类，启动1000000个线程执行入队操作，理论上预期队列大小会小于1000000，但实际输出结果为1000000，想了解原因。

复现代码 #

import threading
import random

class ConcurrentQueue:
    def __init__(self):
        self.queue = []

    def enqueue(self, item):
        self.queue.append(item)

    def size(self):
        return len(self.queue)

NUM_THREADS = 1000000

def main():
    queue = ConcurrentQueue()

    def enqueue_task():
        queue.enqueue(random.randint(1, 1000000))

    # Create and start enqueue threads
    enqueue_threads = [threading.Thread(target=enqueue_task) for _ in range(NUM_THREADS)]
    for thread in enqueue_threads:
        thread.start()

    # Wait for enqueue threads to finish
    for thread in enqueue_threads:
        thread.join()

    print("size:", queue.size())

if __name__ == "__main__":
    main()

原因解析 #

你看到的结果并非偶然，核心在于CPython中list.append()操作的原子性：

• 字节码级别的原子执行：CPython的全局解释器锁（GIL）会确保单个字节码指令的执行不会被线程切换打断。list.append()在CPython中对应单个LIST_APPEND字节码指令，每个线程调用append时，这个指令会完整执行，不会出现中途被其他线程抢占导致的元素丢失。

• 入队逻辑仅包含原子操作：你的enqueue方法只有self.queue.append(item)这一行代码，没有多步操作（比如先判断队列长度再插入、修改多个变量等）。这种情况下，即使没有显式的线程同步机制，每个入队操作也不会互相干扰。

• 注意这是实现细节而非语言规范：这只是CPython的特定实现行为，不属于Python语言标准。如果换成其他Python解释器（如PyPy），或者给enqueue添加多步逻辑（比如if len(self.queue) < limit: self.queue.append(item)），就很可能出现线程安全问题，导致队列大小低于预期。

简言之，你的代码在CPython环境下“碰巧”表现出了线程安全的结果，但这只是依赖了底层实现的原子操作，并不代表ConcurrentQueue类的设计是线程安全的。

内容的提问来源于stack exchange，提问作者nketu06