设计有限阻塞队列

难度: Medium

Submission

运行时间: 34 ms

内存: 17.2 MB

import threading

class BoundedBlockingQueue(object):
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.queue = []
        self.lock = threading.Lock()
        self.full = threading.Semaphore(0)
        self.empty = threading.Semaphore(capacity)

    def enqueue(self, element: int) -> None:
        self.empty.acquire()
        with self.lock:
            self.queue.append(element)
        self.full.release()

    def dequeue(self) -> int:
        self.full.acquire()
        with self.lock:
            element = self.queue.pop(0)
        self.empty.release()
        return element

    def size(self) -> int:
        with self.lock:
            return len(self.queue)

Explain

这个解决方案实现了一个有限阻塞队列，使用 Python 的 threading 模块。主要组件包括一个列表用作队列、一个锁（lock）用于同步访问队列，以及两个信号量（Semaphore）来控制队列的容量。信号量 full 表示队列中的项目数，信号量 empty 表示队列中剩余的空间数。当尝试入队（enqueue）一个元素时，如果队列已满（即 empty 信号量为 0），该操作将阻塞直到队列中有空间。当尝试出队（dequeue）一个元素时，如果队列为空（即 full 信号量为 0），该操作将阻塞直到队列中有元素。这确保了线程安全地访问和修改队列的数据。

时间复杂度: enqueue 和 size 的时间复杂度是 O(1)，dequeue 的时间复杂度是 O(n)。

空间复杂度: O(capacity)

import threading

class BoundedBlockingQueue(object):
    def __init__(self, capacity: int):
        self.capacity = capacity  # 队列的容量限制
        self.queue = []  # 使用列表作为队列的数据结构
        self.lock = threading.Lock()  # 互斥锁保证线程安全的访问
        self.full = threading.Semaphore(0)  # 初始化时没有元素，full 信号量为 0
        self.empty = threading.Semaphore(capacity)  # 初始化时队列为空，empty 信号量等于队列容量

    def enqueue(self, element: int) -> None:
        self.empty.acquire()  # 等待有空位
        with self.lock:  # 在锁的保护下操作队列，保证线程安全
            self.queue.append(element)  # 元素入队
        self.full.release()  # 释放一个 full 信号量，表示队列中有了一个新的元素

    def dequeue(self) -> int:
        self.full.acquire()  # 等待队列中有元素
        with self.lock:  # 在锁的保护下操作队列，保证线程安全
            element = self.queue.pop(0)  # 移除队列的第一个元素
        self.empty.release()  # 释放一个 empty 信号量，表示队列中有了一个空位
        return element  # 返回被移除的元素

    def size(self) -> int:
        with self.lock:  # 在锁的保护下获取队列的大小
            return len(self.queue)  # 返回队列的当前元素数量

Explore

信号量提供了一种简单且有效的方法来控制对有限资源的访问，这在有限阻塞队列的场景中非常适用。使用信号量可以直接控制队列的空间和元素数量，这样可以很直观地实现阻塞和释放线程的行为。当队列满时，入队操作会阻塞，直到信号量表示有空间可用；同理，当队列为空时，出队操作也会阻塞。相比之下，条件变量虽然也能实现相似的功能，但通常需要更复杂的条件判断和锁的管理，可能会使得代码实现更加复杂。信号量的使用在这里提供了清晰和直接的资源控制，减少了出错的可能性。

列表和锁的组合提供了更高的灵活性和控制权。虽然Python的Queue模块提供了一个现成的线程安全队列，它内部已经处理了多线程的竞争条件，但使用基本的数据结构（如列表）和显式的锁可以让开发者更精确地控制锁的粒度和持有时间。这种方式也使得开发者能够更好地理解和学习底层同步机制的工作原理，是一个教学和实验性质的实现。此外，使用列表和锁可以自定义特定的行为，比如在这个场景中与信号量的结合使用，以实现特定的阻塞逻辑。

在这个实现中，信号量控制着访问队列的线程数量——确保不会有超过队列容量的线程执行入队操作，也确保不会有线程在队列为空时执行出队操作。锁（互斥锁）则用于保护队列数据结构的完整性，确保在任何时刻只有一个线程可以修改队列（添加或移除元素）。这样的设计确保了，即使多个线程尝试同时入队或出队，每个操作都是原子的，遵循先入先出的原则，并且队列的数据不会因并发访问而出现脏读或无效状态。

在当前的实现中，如果正确使用，不存在死锁的风险，因为锁的获取和释放顺序是一致和预定的。每次只有一个锁在操作中，且信号量的使用保证了不会有无限等待资源的情况出现。但是，如果未正确处理异常或信号量的释放，可能会导致资源未被正确释放，从而导致阻塞或死锁。为了改进，可以引入异常处理机制，在操作过程中如果出现异常，确保释放所有已获取的锁和信号量，防止资源泄露。此外，监控和日志记录也可用于检测和预防死锁情况，提高系统的稳定性和可靠性。