如何为单一线程锁定队列 连续调用queue.get()时阻塞其他线程

Python标准库queue模块的所有实例自带的内部锁，仅在单次put()/get()操作执行的瞬间持有，操作完成后会立刻释放。你在特殊线程里循环调用get()时，两次调用的间隙锁已经处于释放状态，其他生产者、消费者线程完全可以抢到锁修改队列，无法实现你要的全程阻塞效果。

不要直接修改队列实例的内部锁属性（不同Python版本内部实现不兼容，极易出死锁问题），自行维护一把全局可重入锁threading.RLock，所有线程的所有队列操作，都必须先抢到这把锁才能执行：

• 普通生产者、消费者线程：每次执行put()/get()前先尝试抢锁，抢到后先判断队列是否满足操作条件（不满可写、不空可读），满足就执行单次操作后释放锁；不满足就立刻释放锁，短暂等待后重试，也可以搭配条件变量做唤醒避免空转。绝对不能在持锁状态下做阻塞式的队列操作，否则会直接触发死锁。
• 特殊批量处理线程：到触发周期后直接抢这把全局锁，抢到后持续持有，循环调用get()直到队列清空，再释放锁。整个取数过程中锁始终被当前线程持有，其他所有线程的队列操作都会因为抢不到锁被阻塞，完全不会插入到两次get()的间隙中。取完元素后把业务处理逻辑放到锁外面执行，尽量缩短锁占用时长，避免普通线程长时间排队。

import threading
import queue
import time

# 初始化队列与全局可重入锁
work_queue = queue.Queue(maxsize=20)
queue_access_lock = threading.RLock()

def normal_producer():
    """普通生产者线程逻辑"""
    while True:
        acquired = queue_access_lock.acquire(blocking=False)
        if acquired:
            try:
                if not work_queue.full():
                    work_queue.put(f"prod_item_{time.time()}")
            finally:
                queue_access_lock.release()
        time.sleep(0.005)

def normal_consumer():
    """普通消费者线程逻辑"""
    while True:
        acquired = queue_access_lock.acquire(blocking=False)
        if acquired:
            try:
                if not work_queue.empty():
                    item = work_queue.get()
                    print(f"普通消费者拿到元素: {item}")
            finally:
                queue_access_lock.release()
        time.sleep(0.005)

def special_batch_worker():
    """特殊批量处理线程逻辑"""
    while True:
        time.sleep(3) # 每3秒触发一次全量处理
        print("[特殊线程] 开始锁定队列取全量数据")
        batch = []
        # 抢到锁后全程持有，直到取完所有元素
        with queue_access_lock:
            while not work_queue.empty():
                batch.append(work_queue.get())
        # 锁释放后再做业务处理，不占用锁资源
        print(f"[特殊线程] 取数完成，共拿到{len(batch)}个元素，开始批量处理")
        # 此处写自定义批量处理逻辑即可

# 启动测试线程
if __name__ == "__main__":
    for _ in range(3):
        threading.Thread(target=normal_producer, daemon=True).start()
    for _ in range(2):
        threading.Thread(target=normal_consumer, daemon=True).start()
    threading.Thread(target=special_batch_worker, daemon=True).start()

    while True:
        time.sleep(1)

不要图省事直接操作queue实例的内部mutex、not_empty、not_full属性，这些是内部实现细节，版本变动没有兼容性承诺，线上出问题极难排查。
如果队列吞吐量很高，普通线程用sleep重试的空转开销不可接受，可以把RLock替换为threading.Condition，在队列状态变化、特殊线程释放锁的时候主动notify等待的线程，性能会好很多，核心逻辑还是所有队列操作必须先过这把全局锁。
特殊线程持锁期间只做“把队列元素全部搬运到本地列表”这一件事，所有耗时的业务处理都放到锁释放之后做，尽可能减少锁占用时间。

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