Python基于Lock和Condition实现线程安全队列详解

一、线程池基础组件与同步原理

线程池架构依赖任务调度与数据交换，其中线程安全队列是连接生产端与消费端的核心枢纽。为避免多线程同时读写引发数据竞争，需引入互斥锁控制临界区，并结合条件变量实现高效的阻塞唤醒机制。一个基础的线程池主要由任务对象、线程处理对象、线程池控制器以及本节的线程安全队列构成。

二、线程安全队列核心实现

自定义队列需覆盖添加、批量放入、读取指定索引及弹出元素等操作。为保证线程安全，所有底层列表操作必须包裹在 self.lock 保护下；put 方法需在队列满时拦截并返回异常；pop 方法支持通过 block 参数控制是否挂起当前线程，配合 timeout 参数可防止无限期等待。当新元素入队时，通过条件变量触发通知，解除处于 wait 状态的消费者线程。

class ThreadSafeQueue(object):

    def __init__(self, max_size=0):
        self.queue = []
        self.max_size = max_size  # max_size为0表示无限大
        self.lock = threading.Lock()  # 互斥量
        self.condition = threading.Condition()  # 条件变量

    def size(self):
        """
        获取当前队列的大小
        :return: 队列长度
        """
        # 加锁
        self.lock.acquire()
        size = len(self.queue)
        self.lock.release()
        return size

    def put(self, item):
        """
        将单个元素放入队列
        :param item:
        :return:
        """
        # 队列已满 max_size为0表示无限大
        if self.max_size != 0 and self.size() >= self.max_size:
            return ThreadSafeException()

        # 加锁
        self.lock.acquire()
        self.queue.append(item)
        self.lock.release()
        self.condition.acquire()
        # 通知等待读取的线程
        self.condition.notify()
        self.condition.release()

        return item

    def batch_put(self, item_list):
        """
        批量添加元素
        :param item_list:
        :return:
        """
        if not isinstance(item_list, list):
            item_list = list(item_list)

        res = [self.put(item) for item in item_list]

        return res

    def pop(self, block=False, timeout=0):
        """
        从队列头部取出元素
        :param block: 是否阻塞线程
        :param timeout: 等待时间
        :return:
        """
        if self.size() == 0:
            if block:
                self.condition.acquire()
                self.condition.wait(timeout)
                self.condition.release()
            else:
                return None

        # 加锁
        self.lock.acquire()
        item = None
        if len(self.queue):
            item = self.queue.pop()
        self.lock.release()

        return item

    def get(self, index):
        """
        获取指定位置的元素
        :param index:
        :return:
        """
        if self.size() == 0 or index >= self.size():
            return None

        # 加锁
        self.lock.acquire()
        item = self.queue[index]
        self.lock.release()

        return item


class ThreadSafeException(Exception):
    pass

三、核心功能验证测试

第1步：搭建生产者与消费者模拟环境

创建 ThreadSafeQueue 实例并设置容量上限。编写循环逻辑的生产者线程按固定间隔向队列投递随机整数，消费者线程则调用带阻塞参数的 pop 方法获取数据。此步骤用于观察状态流转与时间戳变化，验证基础通信链路。

第2步：运行阻塞等待与唤醒测试

将 pop 方法的 block 设为 True，timeout 设为 3 秒。当队列为空时，消费者线程自动进入休眠；待生产者插入新元素触发 condition.notify() 后，等待立即解除。执行日志显示的时间差符合预期设定，证明阻塞逻辑与条件变量唤醒机制已正确生效。

第3步：运行多线程序列化访问测试

启动两个独立的消费者线程（thread1、thread2）共享同一队列资源。由于 pop 内部持有了 self.lock，系统会强制排队处理请求。控制台打印结果呈现明显的先后交替顺序，确认了互斥量成功隔离了并发冲突，保证了读写的原子性。

四、关键注意事项与总结

该实现覆盖了并发控制的基础范式，适用于学习 threading 包的底层交互。需注意实际工程中 self.queue.pop() 默认移除末尾元素，若需严格遵循先进先出（FIFO）语义，建议替换为 self.queue.pop(0)。此外，频繁创建自定义队列可能带来额外开销，生产级场景推荐直接调用内置 queue.Queue。以上就是Python实现线程安全队列的详细内容，更多关于Python并发编程的资料请关注本站其它相关文章！