Python queue 包中作为底层结构的 deque 实现开始，到锁和线程安全

在 LeetCode 的 20. 有效的括号 这道题中，需要使用到栈，以及栈的获取头节点的操作，但是 python 的 queue 中的 LifoQueue() 并没有这个操作。只能通过一次“取出回填”的操作才能实现经典的 peek 操作。

import queue

stack = queue.LifoQueue()

def stack_peek(stack):
    if stack.empty(): # 判断不为空
        return None

    val = stack.get() # 取出头值
    stack.put(val) # 放回头值

    # 得到头值
    return val

但是基于 LifoQueue() 实现的方案用时仅打败了 23.46% 的用户。而我们在 “跳”的思想：从 Python deque 到 Redis 跳表，再到 B+/Blink 树 这篇中博客中讨论过，LifoQueue() 是基于 queue.deque 实现的，底层是使用的 双向循环链表 + 定长数组。于是用 deque 重新实现了一次这份代码，这次在用时上，打败了 100% 的用户。

import queue


class Solution:
    def isValid(self, s: str) -> bool:
        # 尾部操作
        # .append(x)
        # .pop()
        # deque[-1]
        # if not deque
        stack = queue.deque()

        for item in s:
            # 当为左半边括号
            if item == "(":
                stack.append(")")
            elif item == "[":
                stack.append("]")
            elif item == "{":
                stack.append("}")
            # 当为右半边括号
            # 栈为空 且 当前元素为右（有元素未处理）
            elif (not stack) and item:
                return False
            # 栈不为空 且 当前元素不匹配
            elif stack and stack[-1] != item:
                return False
            # 栈不为空 且 当前元素匹配
            else:
                stack.pop()

        return not stack

很自然地，我开始对这个包的细节感到好奇。为什么要用 deque 作为基础结构，进而实现其它的数据结构呢？这么做有什么好处？我是不是应该在所有情况下优先使用 deque ？

以下展示 Python 3.12 中队列相关工具的所有可用方法，包括 collections.deque 和 queue 模块的各个类。

🔍 意外发现：queue.deque() 隐藏接口

在深入研究过程中，我们发现了一个有趣的现象：虽然官方文档中没有提及，但是可以通过 queue.deque() 直接访问 collections.deque！

通过查看 Python CPython 源码，我们发现在 queue.py 的第 9 行有 from collections import deque 的导入语句。虽然 deque 没有在模块的 __all__ 列表中声明，但由于 Python 的模块导入机制，它仍然可以通过 queue.deque 访问。

这是一个"隐藏"的可用接口——不在官方文档中推荐，但确实存在且功能完全等同于 collections.deque。虽然可以使用，但建议还是使用标准的 collections.deque 以保持代码的可读性和兼容性。

Python queue 常用方法的生产场景优势劣势

在实际开发中，选择合适的队列工具对程序性能和稳定性至关重要。让我们从实际使用场景的角度来分析这些工具的优势和劣势。

collections.deque：高性能的双端队列

核心优势：

• 极高的两端操作效率：append()、appendleft()、pop()、popleft() 都是 O(1) 时间复杂度
• 内存效率：采用块链式结构，避免了 list 动态扩容时的内存拷贝
• 功能丰富：支持旋转（rotate）、反转（reverse）等特殊操作
• 无锁开销：单线程场景下性能极佳

主要劣势：

• 非线程安全：多线程环境下需要手动加锁
• 随机访问较慢：中间位置的索引访问效率不如 list

生产场景：

• 单线程下的栈和队列操作
• 滑动窗口算法实现
• 历史记录管理（如浏览器历史）

queue.Queue：线程安全的 FIFO 队列

核心优势：

• 线程安全：内置锁机制，多线程下无需额外同步
• 生产者-消费者友好：支持 task_done() 和 join() 进行任务跟踪
• 阻塞控制：支持阻塞和非阻塞操作，便于流量控制

主要劣势：

• 性能开销：锁机制带来额外开销，单线程场景下比 deque 慢
• 功能受限：无法直接查看队列头元素（peek 操作）

生产场景：

• 生产者-消费者模式的数据传递
• 线程池任务分发
• 异步任务队列

queue.LifoQueue：线程安全的栈

核心优势：

• 线程安全的栈操作：多线程环境下的安全 LIFO 实现
• 与 Queue 一致的 API：学习成本低

主要劣势：

• 缺少栈特有功能：没有 peek 操作，无法不弹出就查看栈顶
• 性能开销：同样有锁的开销

生产场景：

• 多线程下的任务调度（后进先处理）
• 表达式求值和语法分析的并行处理

queue.PriorityQueue：线程安全的优先级队列

核心优势：

• 自动优先级排序：基于堆实现，自动维护优先级顺序
• 线程安全：多线程环境下的安全优先级处理

主要劣势：

• 数据格式限制：必须使用 (优先级, 数据) 元组格式
• 堆操作复杂度：入队和出队都是 O(log n)

生产场景：

• 任务调度系统（按优先级处理）
• 多线程爬虫的 URL 优先级管理
• 游戏开发中的事件处理

queue.SimpleQueue：轻量级线程安全队列

核心优势：

• 性能优异：比其他 queue 类更轻量级，开销更小
• 简化 API：去除了复杂的任务跟踪功能，专注核心队列操作
• 无界限制：put() 操作永不阻塞，适合高吞吐场景

主要劣势：

• 功能受限：没有 maxsize、task_done()、join() 等高级功能
• 无流量控制：缺少背压机制，可能导致内存问题

生产场景：

• 高性能消息传递（不需要任务跟踪）
• 简单的线程间通信
• 替代 queue.Queue 的轻量级方案

工具选择指南

这些工具的选择原则：

• 单线程追求性能 → collections.deque
• 多线程需要安全 + 任务跟踪 → queue.Queue / queue.LifoQueue / queue.PriorityQueue
• 多线程需要安全 + 追求性能 → queue.SimpleQueue
• 隐藏接口使用 → queue.deque（等同于 collections.deque，但不推荐）

接下来我们深入了解这些工具的底层实现原理。

Python queue 包常用队列工具的底层实现，以及 deque 和他们的关系

理解"为什么 queue 模块要用 deque 作为底层存储"的关键，在于认识到不同数据结构的设计目标和性能特点。让我们从底层实现的角度来深入分析这个关系。

首先澄清一个概念

需要明确的是：deque 不是所有 Queue 的底层，而是 queue.Queue 和 queue.LifoQueue 的底层存储结构。queue.PriorityQueue 的底层其实是基于 heapq 模块实现的堆结构。

让我们逐一分析：

collections.deque 的底层设计

deque（双端队列）采用了一种巧妙的"块链式结构"（block-linked structure），这种设计平衡了内存效率和操作性能：

核心存储结构

头尾元素操作（核心优势）

批量操作与特殊功能

queue 模块如何利用 deque

queue 模块的核心目标是提供线程安全的队列操作。它采用了"分层设计"的思路：

• 存储层：使用 deque 或 heapq 提供高效的数据操作
• 同步层：使用 threading.Lock 和 threading.Condition 提供线程安全保障

queue.Queue：FIFO 队列的实现

为什么选择 deque？

• FIFO 队列需要"尾部入队，头部出队"的操作模式
• deque.append() + deque.popleft() 都是 O(1) 操作，完美匹配需求
• 相比 list，避免了头部删除时的 O(n) 元素移动开销

queue.LifoQueue：栈的实现

为什么也用 deque？

• 栈需要"尾部入栈，尾部出栈"的操作模式
• deque.append() + deque.pop() 同样是 O(1) 操作
• 保持与 Queue 一致的底层结构，便于代码维护

queue.PriorityQueue：优先级队列的例外

为什么不用 deque？

• 优先级队列需要自动排序功能，每次出队都要取出最高优先级元素
• deque 无法提供排序功能，而 heapq 模块的堆结构天然支持优先级
• 使用 list 作为堆的容器，通过 heapq.heappush()/heapq.heappop() 维护优先级顺序

工具对比总结

这种设计哲学体现了一个重要原则：底层结构的选择，始终服务于上层功能需求。deque 提供了高效的双端操作基础，queue 模块在此基础上添加了线程安全的保障，形成了完整的多线程队列解决方案。

但这里就引出了一个关键问题：queue 模块是如何实现线程安全的？这就需要我们深入了解 Python 的锁机制。

线程锁的源码和底层实现

要理解 Python 3.12 中 queue.Queue（基于 deque + 锁）的锁实现逻辑，需从两个核心层面拆解：

1. queue.Queue 自身使用的锁类型（threading.Lock/threading.Condition）；
2. 底层锁的源码实现（Python 解释器层面，分「纯 Python 封装」和「C 语言核心」）。

queue.Queue 使用了哪些锁？

queue.Queue 并非仅用一个简单锁，而是基于 threading.Condition（条件变量）实现，而 Condition 又依赖 threading.Lock（互斥锁）。其核心逻辑是：

• 底层存储：用 collections.deque 保存队列数据（deque 的 append/popleft 本身是原子操作，但 Queue 需额外控制「空队列阻塞取」「满队列阻塞存」，因此需要锁）；
• 锁机制：通过 Condition 实现「线程间同步」——既保证 deque 操作的原子性（互斥），又支持「等待/唤醒」逻辑（如队空时 get() 阻塞，有数据时被 put() 唤醒）。

我们先看 queue.Queue 源码中锁的初始化（Python 3.12 标准库 queue.py）：

class Queue:
    def __init__(self, maxsize=0):
        self.maxsize = maxsize
        self._init(maxsize)  # 实际调用 _deque = deque()
        # 关键：创建 Condition 对象（内部包含一个 Lock）
        self.mutex = threading.Lock()  # 互斥锁，保护队列状态
        self.not_empty = threading.Condition(self.mutex)  # 条件变量：队列非空
        self.not_full = threading.Condition(self.mutex)   # 条件变量：队列未满
        # ... 其他属性 ...

    def _init(self, maxsize):
        self._deque = deque()  # 底层存储：deque

可见：Queue 的锁核心是 threading.Condition，而 Condition 又封装了 threading.Lock。要理解锁的实现，本质是理解 threading.Lock 和 threading.Condition 的源码。

threading.Lock：底层互斥锁的实现

threading.Lock 是 Python 多线程的基础互斥锁，保证同一时间只有一个线程能进入「临界区」。其实现分两层：

• Python 层面：Lock 类是对底层 C 实现的「封装」（暴露 acquire()/release() 等接口）；
• C 语言层面：依赖操作系统的「原生互斥锁」（如 Linux 的 pthread_mutex_t、Windows 的 CRITICAL_SECTION），保证跨平台一致性。

Python 层面：Lock 类的封装

Python 3.12 中 threading.Lock 的定义非常简洁，核心是调用底层 C 模块 _thread（CPython 内置模块，封装操作系统锁）：

class Lock:
    def __init__(self):
        # 调用 _thread.allocate_lock() 创建底层锁对象
        self._lock = _thread.allocate_lock()
        # ... 其他属性（如所有者、递归计数，仅 RLock 用）...

    def acquire(self, blocking=True, timeout=-1):
        # 调用底层锁的 acquire 方法，支持阻塞/超时
        return self._lock.acquire(blocking, timeout)

    def release(self):
        # 调用底层锁的 release 方法
        self._lock.release()

    def locked(self):
        # 判断锁是否已被持有
        return self._lock.locked()

    # ... 其他方法（如 __enter__/__exit__ 支持 with 语句）...

可见：Python 层面的 Lock 只是「壳」，真正的锁逻辑在 _thread.allocate_lock() 创建的对象中——而 _thread 是 CPython 的 C 扩展模块。

C 语言层面：_thread 模块的锁实现

CPython 的 _thread 模块（源码路径：Modules/_threadmodule.c）直接调用操作系统原生的互斥锁 API，实现跨平台的线程互斥。核心逻辑如下：

锁对象的结构体定义

C 层面用 thread_lock 结构体表示一个锁，内部包含操作系统的锁句柄（如 Linux 的 pthread_mutex_t、Windows 的 CRITICAL_SECTION）：

typedef struct {
    #ifdef MS_WINDOWS
        CRITICAL_SECTION cs;  // Windows 临界区（轻量级互斥锁）
        int owned;            // 标记是否被当前线程持有
    #else
        pthread_mutex_t mutex;  // POSIX 互斥锁（Linux/macOS）
    #endif
    // ... 其他状态（如是否递归锁、超时控制）...
} thread_lock;

锁的创建（allocate_lock 逻辑）

当调用 _thread.allocate_lock() 时，C 代码会初始化操作系统的锁句柄：

• Windows 平台：调用 InitializeCriticalSection(&lock->cs) 初始化临界区；
• POSIX 平台（Linux/macOS）：调用 pthread_mutex_init(&lock->mutex, NULL) 初始化互斥锁（默认属性：非递归、阻塞）。

关键代码片段（简化）：

static PyObject *
thread_allocate_lock(PyObject *self, PyObject *args) {
    thread_lock *lock = PyMem_Malloc(sizeof(thread_lock));
    if (lock == NULL) return PyErr_NoMemory();

    #ifdef MS_WINDOWS
        InitializeCriticalSection(&lock->cs);
        lock->owned = 0;
    #else
        if (pthread_mutex_init(&lock->mutex, NULL) != 0) {
            PyMem_Free(lock);
            return PyErr_SetFromErrno(PyExc_OSError);
        }
    #endif

    // 将 C 结构体包装成 Python 对象返回
    return (PyObject *)_thread_lock_new(lock);
}

锁的获取（acquire 逻辑）

当调用 lock.acquire() 时，C 代码会调用操作系统的「锁获取 API」：

• Windows：EnterCriticalSection(&lock->cs)（阻塞直到获取锁）；若指定 blocking=False，则用 TryEnterCriticalSection（非阻塞，获取失败直接返回）。
• POSIX：pthread_mutex_lock(&lock->mutex)（阻塞）；非阻塞时用 pthread_mutex_trylock，超时控制用 pthread_mutex_timedlock。

锁的释放（release 逻辑）

调用 lock.release() 时，C 代码调用操作系统的「锁释放 API」：

• Windows：LeaveCriticalSection(&lock->cs)；
• POSIX：pthread_mutex_unlock(&lock->mutex)。

若锁未被当前线程持有却调用 release()，会触发「解锁未持有锁」的错误（Python 层面抛出 RuntimeError）。

threading.Condition：条件变量的实现

queue.Queue 中 not_empty/not_full 是 Condition 对象，它基于 Lock 实现「线程间等待/唤醒」——解决「队列空时不能取」「队列满时不能存」的问题。其核心逻辑是：

• 互斥基础：Condition 内部持有一个 Lock（可传入自定义 Lock，默认新建一个），所有对共享资源（如 deque）的操作必须先通过 Condition 获取 Lock；
• 等待/唤醒：通过「条件变量」实现线程阻塞（wait()）和唤醒（notify()/notify_all()），避免线程「忙轮询」（反复检查队列状态，浪费 CPU）。

Python 层面：Condition 类的核心逻辑

class Condition:
    def __init__(self, lock=None):
        # 若未传入锁，默认创建一个 Lock 对象
        self._lock = lock or Lock()
        # 底层用 _thread.Condition（C 实现），依赖 self._lock 的 C 锁句柄
        self._cond = _thread.allocate_condition()

    def acquire(self, *args, **kwargs):
        # 获取内部锁（进入临界区）
        return self._lock.acquire(*args, **kwargs)

    def release(self):
        # 释放内部锁（退出临界区）
        self._lock.release()

    def wait(self, timeout=-1):
        # 1. 释放内部锁（让其他线程能进入临界区）
        # 2. 阻塞当前线程，等待被 notify() 唤醒
        # 3. 唤醒后重新获取内部锁
        with self._lock:
            return self._cond.wait(timeout)

    def notify(self, n=1):
        # 唤醒 n 个等待在该条件变量上的线程
        self._cond.notify(n)

    def notify_all(self):
        # 唤醒所有等待的线程
        self._cond.notify_all()

    # ... 支持 with 语句（__enter__ = acquire，__exit__ = release）...

C 层面：_thread.Condition 的实现

_thread.allocate_condition() 创建的条件变量，同样依赖操作系统原生 API：

• Windows：用 HANDLE（事件对象 CreateEvent）实现等待/唤醒；
• POSIX：用 pthread_cond_t（POSIX 条件变量）实现，需与 pthread_mutex_t（Lock 的底层锁）配合使用（保证等待/唤醒的原子性）。

核心逻辑：wait() 会先释放锁、阻塞线程，被唤醒后重新获取锁；notify() 会唤醒阻塞的线程，但需等待线程重新获取锁后才能继续执行。

queue.Queue 的锁实现链路

Python 3.12 中 queue.Queue = deque + 锁 的完整链路如下：

queue.Queue
├─ 底层存储：collections.deque（原子操作基础）
└─ 锁机制：threading.Condition（not_empty / not_full）
   ├─ 依赖：threading.Lock（互斥核心）
   │  ├─ Python 层面：Lock 类封装（暴露 acquire/release）
   │  └─ C 层面：_thread 模块调用操作系统原生锁
   │     ├─ Windows：CRITICAL_SECTION（临界区）
   │     └─ POSIX（Linux/macOS）：pthread_mutex_t（互斥锁）
   └─ 条件变量：_thread.Condition（等待/唤醒）
      ├─ Windows：Event（事件对象）
      └─ POSIX：pthread_cond_t（条件变量）

本质上，Python 的线程锁并非「自己实现」，而是封装了操作系统的原生锁 API——因为线程是操作系统级别的调度单位，只有操作系统能保证不同线程间的互斥和同步。Python 做的只是提供统一的、跨平台的 API 接口，让开发者无需直接操作操作系统底层。

理解了锁的实现原理，接下来我们就能更好地理解什么是线程安全了。

什么是线程安全，什么是多线程安全

在 Python 及其他编程语言的并发编程中，线程安全（Thread Safety） 和 多线程安全（Multi-thread Safety） 本质是同一概念的不同表述——指多个线程同时访问共享资源时，程序仍能保证逻辑正确、数据一致，且不会出现未定义行为的特性。核心是解决"多线程竞争共享资源"可能引发的问题，我们可以从"问题根源""核心判断标准""常见场景"三个维度深入理解：

为什么需要"线程安全"？

多线程的核心价值是"并发执行任务，提高资源利用率"，但线程间会共享进程的内存空间（如全局变量、类属性、堆内存中的对象等）。当多个线程同时对共享资源执行"读+写"操作时，会触发"线程竞争（Race Condition）"，导致数据错乱或逻辑异常。

举个非线程安全的例子（Python 代码）：

import threading

# 共享资源：计数器
count = 0

# 线程任务：对计数器加1（执行10000次）
def add_count():
    global count
    for _ in range(10000):
        count += 1  # 看似1步，实际是"读count→加1→写回count"3步操作

# 启动2个线程并发执行
t1 = threading.Thread(target=add_count)
t2 = threading.Thread(target=add_count)
t1.start()
t2.start()
t1.join()
t2.join()

print(count)  # 预期20000，实际常小于20000（如15678、18901等）

问题原因：线程1读取 count=100 后，还没来得及写回，线程2就已读取到同样的 count=100；两者分别加1后写回，最终 count=101（而非预期的102），导致数据错乱。

线程安全的核心判断标准

一个程序/函数/数据结构是否线程安全，关键看它在"多线程并发访问共享资源"时是否满足以下3个条件：

1. 原子性（Atomicity）

对共享资源的操作必须"要么全执行，要么全不执行"，中间不能被其他线程打断。比如上述 count +=1 不是原子操作，而 Python 中 queue.Queue 的 put()/get() 是原子操作（底层通过锁保证）。

2. 可见性（Visibility）

一个线程对共享资源的修改，能被其他线程"及时看到"。若缺乏可见性，线程1修改了数据，但线程2仍读取到旧值（因 CPU 缓存、编译器优化等导致），会引发逻辑错误。

3. 有序性（Ordering）

程序执行的顺序不能被编译器或 CPU 优化打乱（即"指令重排"）。若有序性被破坏，多线程执行的逻辑会偏离预期（如"初始化操作"被排到"使用操作"之后）。

常见的"线程安全"与"非线程安全"场景

线程安全的工具/操作

• queue.Queue 类：Python 标准库为多线程设计的队列，底层通过 threading.Lock 保证 put()/get() 等操作的原子性，支持多线程安全地生产/消费数据。
• collections.deque 的部分操作：deque 是线程安全的"双端队列"，其 append()/appendleft()/pop()/popleft() 是原子操作（但需注意：若组合多个操作，如"先判断非空再 pop"，仍需额外加锁，因为"判断+pop"整体不是原子的）。
• threading.Lock/RLock：手动加锁工具，通过"互斥"保证同一时间只有一个线程能进入"临界区"（访问共享资源的代码块），本质是人为实现线程安全。
• 原子操作：如 Python 中对整数的 += 若用 ctypes 或 concurrent.futures.thread 中的原子工具，或简单的"读操作"（仅读取共享资源，不修改），天然线程安全。

非线程安全的场景

• 普通全局变量/列表/字典：如 list.append() 本身是原子操作（Python 底层实现保证），但 list = list + [1] 不是（需"读列表→拼接→写回列表"）；字典的 dict[key] = value 也不是原子操作（若多个线程同时修改同一 key，可能导致哈希表错乱）。
• 自定义类的属性：若类属性被多个线程同时修改（如 class A: x=0，多线程执行 A.x +=1），会因 += 非原子性导致数据错乱。
• 无锁的文件操作：多个线程同时读写同一个文件，会导致文件内容错乱（如线程1写一半，线程2插入内容）。

如何实现线程安全？

当遇到非线程安全的场景时，可通过以下3种核心方式保证线程安全：

1. 使用线程安全的数据结构

优先选择标准库提供的线程安全工具，如 queue.Queue 替代自定义队列，deque 替代普通列表（仅用其原子操作）。

2. 手动加锁（互斥锁）

用 threading.Lock 包裹"访问共享资源的代码块"，保证同一时间只有一个线程执行：

import threading

count = 0
lock = threading.Lock()  # 创建互斥锁

def add_count():
    global count
    for _ in range(10000):
        with lock:  # 自动加锁/释放锁（避免手动释放遗漏）
            count += 1  # 临界区：仅一个线程能进入

# 启动线程后，count 最终会等于 20000（线程安全）

3. 避免共享资源

通过"线程本地存储（Thread-Local Storage）"让每个线程持有独立的数据，无需共享。如 threading.local() 创建的对象，每个线程访问时都是独立的副本，天然无竞争：

import threading

local_data = threading.local()  # 线程本地存储

def set_local():
    local_data.x = 1  # 每个线程的 x 是独立的，不共享

t1 = threading.Thread(target=set_local)
t2 = threading.Thread(target=set_local)
# 线程1的 local_data.x 和线程2的互不影响，无需锁

总结：从 LeetCode 到生产实践

回到我们开始的话题，从 LeetCode 20. 有效的括号这道题开始，我们经历了：

1. 性能问题发现：LifoQueue() 比 deque 慢的现象
2. 工具选择学习：理解不同队列工具的适用场景和性能特点
3. 底层原理探索：深入了解 deque 和 queue 的实现机制
4. 锁机制剖析：掌握 Python 线程锁的源码和原理
5. 线程安全认知：明确多线程编程的核心概念和实践方法

这个学习过程体现了一个重要的认知：技术选择永远要服务于实际场景需求。

• 单线程追求性能：选择 deque，享受 O(1) 的高效操作
• 多线程需要安全：选择 queue 模块，获得可靠的并发保障
• 理解底层原理：帮助我们在复杂场景下做出正确的技术决策

在 Python 中，因 GIL（全局解释器锁）的存在，多线程在 CPU 密集型任务中并发效率有限，但"线程安全"依然重要——GIL 仅保证"同一时间一个线程执行字节码"，但无法避免"多线程对共享资源的非原子操作"（如上述 count +=1 问题），仍需通过锁或线程安全工具解决。

最终建议：在实际开发中，优先使用经过验证的线程安全工具，当需要自己实现并发控制时，一定要深入理解锁的机制和限制，确保程序的正确性和性能。