阻塞队列 | 有界缓冲区 | 生产者-消费者
经典的同步问题，涉及一个有限容量的缓冲区，该缓冲区可由不同的生产者线程添加项目或由消费者线程移除项目。此问题有多种名称：生产者-消费者问题、有界缓冲区问题或阻塞队列问题。
阻塞队列 | 有界缓冲区 | 生产者-消费者问题
阻塞队列是指当队列没有足够空间容纳新元素时，调用入队（enqueue）方法的线程会被阻塞。同样地，当队列为空时，调用出队（dequeue）方法的线程也会被阻塞。此外，当队列中有空闲空间时，系统会通知被阻塞的入队线程；当队列中有新元素可用时，系统会通知被阻塞的出队线程。
解决方案
queue.Queue 模块已经内置了线程安全的队列。我们将从零开始实现一个可以被多个生产者和消费者共享的阻塞队列。
我们的队列将具有固定的大小，该大小通过构造函数传入。我们还将使用列表作为底层数据结构来实现队列。可以通过列表的 pop() 和 append() 方法模拟队列的操作。其中 append() 将元素添加到列表末尾，pop(0) 从列表开头移除元素，这样就不需要手动维护队列的头尾指针。
我们需要两个变量：一个用于记录队列的最大容量，另一个用于记录当前队列的大小。此外，还需要一个条件变量，供生产者或消费者在队列满或空时等待使用。
初始类结构如下：
1class BlockingQueue: 2 3 def __init__(self, max_size): 4 self.max_size = max_size 5 self.curr_size = 0 6 self.cond = Condition() 7 self.q = [] 8 9 def enqueue(self): 10 pass 11 12 def dequeue(self): 13 pass
首先实现 enqueue() 方法，该方法接收一个待入队的元素。如果当前队列大小等于最大容量，则需要阻塞调用该方法的线程。可以通过在条件变量上使用 wait() 方法并配合 while 循环来实现。循环条件为队列大小等于最大容量。一旦其他线程执行出队操作，条件将变为假，从而退出循环。
需要注意的是，每次检查 curr_size 变量时，都必须确保没有其他线程正在修改它。这可以通过在每个方法开始时获取与条件变量关联的锁来实现。
入队方法的实现如下：
1 def enqueue(self, item): 2 3 self.cond.acquire() 4 while self.curr_size == self.max_size: 5 self.cond.wait() 6 7 self.q.append(item) 8 self.curr_size += 1 9 10 self.cond.notifyAll() 11 self.cond.release()
最后我们调用了 notifyAll()。由于我们刚刚向队列中添加了一个元素，可能有消费者线程正阻塞在 dequeue() 方法中等待元素出现。因此必须发出信号唤醒所有等待的线程。
如果没有线程在等待，该信号将被忽略，不会影响程序的正确性。这种情况称为 漏信号（missed signal）。
接下来设计 dequeue() 方法。与 enqueue() 类似，当队列为空（即 self.curr_size == 0）时，应阻塞调用出队方法的线程。此时消费者线程需等待条件变量。
消费完一个元素后，记得调用 notifyAll()，因为如果队列已满，可能有生产者线程正阻塞在 enqueue() 方法中等待空间。代码逻辑如下：
1 def dequeue(self): 2 3 self.cond.acquire() 4 while self.curr_size == 0: 5 self.cond.wait() 6 7 item = self.q.pop(0) 8 self.curr_size -= 1 9 10 self.cond.notifyAll() 11 self.cond.release() 12 13 return item
可以看出，出队方法与入队方法非常相似。完整的代码如下所示，并附带示例。
1from threading import Thread 2from threading import Condition 3from threading import current_thread 4import time 5import random 6 7 8class BlockingQueue: 9 10 def __init__(self, max_size): 11 self.max_size = max_size 12 self.curr_size = 0 13 self.cond = Condition() 14 self.q = [] 15 16 def dequeue(self): 17 18 self.cond.acquire() 19 while self.curr_size == 0: 20 self.cond.wait() 21 22 item = self.q.pop(0) 23 self.curr_size -= 1 24 25 self.cond.notifyAll() 26 self.cond.release() 27 28 return item 29 30 def enqueue(self, item): 31 32 self.cond.acquire() 33 while self.curr_size == self.max_size: 34 self.cond.wait() 35 36 self.q.append(item) 37 self.curr_size += 1 38 39 self.cond.notifyAll() 40 print("\ncurrent size of queue {0}".format(self.curr_size), flush=True) 41 self.cond.release() 42 43 44def consumer_thread(q): 45 while 1: 46 item = q.dequeue() 47 print("\n{0} consumed item {1}".format(current_thread().getName(), item), flush=True) 48 time.sleep(random.randint(1, 3)) 49 50 51def producer_thread(q, val): 52 item = val 53 while 1: 54 q.enqueue(item) 55 item += 1 56 time.sleep(0.1) 57 58 59if __name__ == "__main__": 60 blocking_q = BlockingQueue(5) 61 62 consumerThread1 = Thread(target=consumer_thread, name="consumer-1", args=(blocking_q,), daemon=True) 63 consumerThread2 = Thread(target=consumer_thread, name="consumer-2", args=(blocking_q,), daemon=True) 64 producerThread1 = Thread(target=producer_thread, name="producer-1", args=(blocking_q, 1), daemon=True) 65 producerThread2 = Thread(target=producer_thread, name="producer-2", args=(blocking_q, 100), daemon=True) 66 67 consumerThread1.start() 68 consumerThread2.start() 69 producerThread1.start() 70 producerThread2.start() 71 72 time.sleep(15) 73 print("Main thread exiting")
在上述示例中，我们有两个生产者和两个消费者。我们故意通过 sleep 减慢消费者的处理速度。整个模拟运行 15 秒。其中一个生产者从 1 开始向队列中添加数值，另一个从 100 开始。从输出结果可以看出，消费者线程能够接收到两个生产者产生的数据，并且队列的当前大小从未超过设定的最大值。
延伸问题
在我们的实现中，使用 notify() 还是 notifyAll() 有区别吗？
在 enqueue() 和 dequeue() 方法中，我们都使用了 notifyAll() 而不是 notify()。这个选择非常重要。考虑这样一个场景：有两个生产者线程和一个消费者线程，它们共同操作一个容量为 1 的队列。当其中一个生产者线程向队列中添加元素后，另外两个线程可能都处于等待状态。如果此时生产者调用 notify()，系统可能唤醒另一个生产者线程。而这个被唤醒的生产者发现队列已满，又会重新进入等待状态，从而导致死锁。而使用 notifyAll() 可以确保消费者线程也有机会被唤醒并继续执行。