时间:2021-11-09 10:29 作者:admin

我们将使用python/' target='_blank'>python线程来解决Python中的生产者—消费者问题。这个问题完全不像他们在学校中说的那么难。

如果你对生产者—消费者问题有了解，看这篇博客会更有意义。

为什么要关心生产者—消费者问题：

可以帮你更好地理解并发和不同概念的并发。 信息队列中的实现中，一定程度上使用了生产者—消费者问题的概念，而你某些时候必然会用到消息队列。

当我们在使用线程时，你可以学习以下的线程概念：

Condition：线程中的条件。 wait()：在条件实例中可用的wait()。 notify() ：在条件实例中可用的notify()。

我假设你已经有这些基本概念：线程、竞态条件，以及如何解决静态条件（例如使用lock)。否则的话，你建议你去看我上一篇文章basics of Threads。

引用维基百科：

生产者的工作是产生一块数据，放到buffer中，如此循环。与此同时，消费者在消耗这些数据（例如从buffer中把它们移除），每次一块。

这里的关键词是“同时”。所以生产者和消费者是并发运行的，我们需要对生产者和消费者做线程分离。

from threading import Thread class ProducerThread(Thread):  def run(self):    pass class ConsumerThread(Thread):  def run(self):    pass

再次引用维基百科：

这个为描述了两个共享固定大小缓冲队列的进程，即生产者和消费者。

假设我们有一个全局变量，可以被生产者和消费者线程修改。生产者产生数据并把它加入到队列。消费者消耗这些数据（例如把它移出）。

queue = []

在刚开始，我们不会设置固定大小的条件，而在实际运行时加入（指下述例子）。

一开始带bug的程序：

from threading import Thread, Lockimport timeimport random queue = []lock = Lock() class ProducerThread(Thread):  def run(self):    nums = range(5) #Will create the list [0, 1, 2, 3, 4]    global queue    while True:      num = random.choice(nums) #Selects a random number from list [0, 1, 2, 3, 4]      lock.acquire()      queue.append(num)      print "Produced", num      lock.release()      time.sleep(random.random()) class ConsumerThread(Thread):  def run(self):    global queue    while True:      lock.acquire()      if not queue:        print "Nothing in queue, but consumer will try to consume"      num = queue.pop(0)      print "Consumed", num      lock.release()      time.sleep(random.random()) ProducerThread().start()ConsumerThread().start()

运行几次并留意一下结果。如果程序在IndexError异常后并没有自动结束，用Ctrl+Z结束运行。

样例输出：

Produced 3Consumed 3Produced 4Consumed 4Produced 1Consumed 1Nothing in queue, but consumer will try to consumeException in thread Thread-2:Traceback (most recent call last): File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner  self.run() File "producer_consumer.py", line 31, in run  num = queue.pop(0)IndexError: pop from empty list

解释：

我们开始了一个生产者线程（下称生产者）和一个消费者线程（下称消费者）。 生产者不停地添加（数据）到队列，而消费者不停地消耗。 由于队列是一个共享变量，我们把它放到lock程序块内，以防发生竞态条件。 在某一时间点，消费者把所有东西消耗完毕而生产者还在挂起（sleep）。消费者尝试继续进行消耗，但此时队列为空，出现IndexError异常。 在每次运行过程中，在发生IndexError异常之前，你会看到print语句输出”Nothing in queue, but consumer will try to consume”，这是你出错的原因。

我们把这个实现作为错误行为（wrong behavior）。

什么是正确行为？

当队列中没有任何数据的时候，消费者应该停止运行并等待(wait），而不是继续尝试进行消耗。而当生产者在队列中加入数据之后，应该有一个渠道去告诉（notify）消费者。然后消费者可以再次从队列中进行消耗，而IndexError不再出现。

关于条件

条件（condition）可以让一个或多个线程进入wait，直到被其他线程notify。参考：?http://docs.python.org/2/library/threading.html#condition-objects

这就是我们所需要的。我们希望消费者在队列为空的时候wait，只有在被生产者notify后恢复。生产者只有在往队列中加入数据后进行notify。因此在生产者notify后，可以确保队列非空，因此消费者消费时不会出现异常。

condition内含lock。 condition有acquire()和release()方法，用以调用内部的lock的对应方法。

condition的acquire()和release()方法内部调用了lock的acquire()和release()。所以我们可以用condiction实例取代lock实例，但lock的行为不会改变。
生产者和消费者需要使用同一个condition实例, 保证wait和notify正常工作。

重写消费者代码：

from threading import Condition condition = Condition() class ConsumerThread(Thread):  def run(self):    global queue    while True:      condition.acquire()      if not queue:        print "Nothing in queue, consumer is waiting"        condition.wait()        print "Producer added something to queue and notified the consumer"      num = queue.pop(0)      print "Consumed", num      condition.release()      time.sleep(random.random())

重写生产者代码：

class ProducerThread(Thread):  def run(self):    nums = range(5)    global queue    while True:      condition.acquire()      num = random.choice(nums)      queue.append(num)      print "Produced", num      condition.notify()      condition.release()      time.sleep(random.random())

样例输出：

Produced 3Consumed 3Produced 1Consumed 1Produced 4Consumed 4Produced 3Consumed 3Nothing in queue, consumer is waitingProduced 2Producer added something to queue and notified the consumerConsumed 2Nothing in queue, consumer is waitingProduced 2Producer added something to queue and notified the consumerConsumed 2Nothing in queue, consumer is waitingProduced 3Producer added something to queue and notified the consumerConsumed 3Produced 4Consumed 4Produced 1Consumed 1

解释：

对于消费者，在消费前检查队列是否为空。 如果为空，调用condition实例的wait()方法。 消费者进入wait()，同时释放所持有的lock。 除非被notify，否则它不会运行。 生产者可以acquire这个lock，因为它已经被消费者release。 当调用了condition的notify()方法后，消费者被唤醒，但唤醒不意味着它可以开始运行。 notify()并不释放lock，调用notify()后，lock依然被生产者所持有。 生产者通过condition.release()显式释放lock。 消费者再次开始运行，现在它可以得到队列中的数据而不会出现IndexError异常。

为队列增加大小限制

生产者不能向一个满队列继续加入数据。

它可以用以下方式来实现：

在加入数据前，生产者检查队列是否为满。 如果不为满，生产者可以继续正常流程。 如果为满，生产者必须等待，调用condition实例的wait()。 消费者可以运行。消费者消耗队列，并产生一个空余位置。 然后消费者notify生产者。 当消费者释放lock，消费者可以acquire这个lock然后往队列中加入数据。

最终程序如下：

from threading import Thread, Conditionimport timeimport random queue = []MAX_NUM = 10condition = Condition() class ProducerThread(Thread):  def run(self):    nums = range(5)    global queue    while True:      condition.acquire()      if len(queue) == MAX_NUM:        print "Queue full, producer is waiting"        condition.wait()        print "Space in queue, Consumer notified the producer"      num = random.choice(nums)      queue.append(num)      print "Produced", num      condition.notify()      condition.release()      time.sleep(random.random()) class ConsumerThread(Thread):  def run(self):    global queue    while True:      condition.acquire()      if not queue:        print "Nothing in queue, consumer is waiting"        condition.wait()        print "Producer added something to queue and notified the consumer"      num = queue.pop(0)      print "Consumed", num      condition.notify()      condition.release()      time.sleep(random.random()) ProducerThread().start()ConsumerThread().start()

样例输出：

Produced 0Consumed 0Produced 0Produced 4Consumed 0Consumed 4Nothing in queue, consumer is waitingProduced 4Producer added something to queue and notified the consumerConsumed 4Produced 3Produced 2Consumed 3

更新：
很多网友建议我在lock和condition下使用Queue来代替使用list。我同意这种做法，但我的目的是展示Condition，wait()和notify()如何工作，所以使用了list。

以下用Queue来更新一下代码。

Queue封装了Condition的行为，如wait()，notify()，acquire()。

现在不失为一个好机会读一下Queue的文档（http://docs.python.org/2/library/queue.html）。

更新程序：

from threading import Threadimport timeimport randomfrom Queue import Queue queue = Queue(10) class ProducerThread(Thread):  def run(self):    nums = range(5)    global queue    while True:      num = random.choice(nums)      queue.put(num)      print "Produced", num      time.sleep(random.random()) class ConsumerThread(Thread):  def run(self):    global queue    while True:      num = queue.get()      queue.task_done()      print "Consumed", num      time.sleep(random.random()) ProducerThread().start()ConsumerThread().start()

解释：

在原来使用list的位置，改为使用Queue实例（下称队列）。 这个队列有一个condition，它有自己的lock。如果你使用Queue，你不需要为condition和lock而烦恼。 生产者调用队列的put方法来插入数据。 put()在插入数据前有一个获取lock的逻辑。 同时，put()也会检查队列是否已满。如果已满，它会在内部调用wait()，生产者开始等待。 消费者使用get方法。 get()从队列中移出数据前会获取lock。 get()会检查队列是否为空，如果为空，消费者进入等待状态。 get()和put()都有适当的notify()。现在就去看Queue的源码吧。 (责任编辑：admin)