用Python的线程来解决生产者消费问题的示例
|
我们将使用Python线程来解决Python中的生产者―消费者问题。这个问题完全不像他们在学校中说的那么难。 如果你对生产者―消费者问题有了解，看这篇博客会更有意义。 为什么要关心生产者―消费者问题：
当我们在使用线程时，你可以学习以下的线程概念：
我假设你已经有这些基本概念：线程、竞态条件，以及如何解决静态条件（例如使用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,Lock
import time
import 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,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 3 Consumed 3 Produced 4 Consumed 4 Produced 1 Consumed 1 Nothing in queue,but consumer will try to consume Exception 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 解释：
我们把这个实现作为错误行为（wrong behavior）。 什么是正确行为？ 当队列中没有任何数据的时候，消费者应该停止运行并等待(wait），而不是继续尝试进行消耗。而当生产者在队列中加入数据之后，应该有一个渠道去告诉（notify）消费者。然后消费者可以再次从队列中进行消耗，而IndexError不再出现。 关于条件 条件（condition）可以让一个或多个线程进入wait，直到被其他线程notify。参考：?http://docs.python.org/2/library/threading.html#condition-objects 这就是我们所需要的。我们希望消费者在队列为空的时候wait，只有在被生产者notify后恢复。生产者只有在往队列中加入数据后进行notify。因此在生产者notify后，可以确保队列非空，因此消费者消费时不会出现异常。
condition的acquire()和release()方法内部调用了lock的acquire()和release()。所以我们可以用condiction实例取代lock实例，但lock的行为不会改变。 重写消费者代码：
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 3 Consumed 3 Produced 1 Consumed 1 Produced 4 Consumed 4 Produced 3 Consumed 3 Nothing in queue,consumer is waiting Produced 2 Producer added something to queue and notified the consumer Consumed 2 Nothing in queue,consumer is waiting Produced 3 Producer added something to queue and notified the consumer Consumed 3 Produced 4 Consumed 4 Produced 1 Consumed 1 解释：
为队列增加大小限制 生产者不能向一个满队列继续加入数据。 它可以用以下方式来实现：
最终程序如下：
from threading import Thread,Condition
import time
import random
queue = []
MAX_NUM = 10
condition = 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,num
condition.notify()
condition.release()
time.sleep(random.random())
ProducerThread().start()
ConsumerThread().start()
（编辑：安卓应用网） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |