用Python的线程来解决生产者消费问题的示例_python

我们将使用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, 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, 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 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

解释:

  •     我们开始了一个生产者线程(下称生产者)和一个消费者线程(下称消费者)。
  •     生产者不停地添加(数据)到队列,而消费者不停地消耗。
  •     由于队列是一个共享变量,我们把它放到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 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 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

解释:

  •     对于消费者,在消费前检查队列是否为空。
  •     如果为空,调用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, 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, 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 0
Consumed 0
Produced 0
Produced 4
Consumed 0
Consumed 4
Nothing in queue, consumer is waiting
Produced 4
Producer added something to queue and notified the consumer
Consumed 4
Produced 3
Produced 2
Consumed 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 Thread
import time
import random
from 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的源码吧。

以上是小编为您精心准备的的内容,在的博客、问答、公众号、人物、课程等栏目也有的相关内容,欢迎继续使用右上角搜索按钮进行搜索python
线程
多线程生产者消费者、java线程生产者消费者、线程 生产者消费者、线程池 生产者消费者、c 线程 生产者消费者,以便于您获取更多的相关知识。

时间: 2024-12-03 04:20:39

用Python的线程来解决生产者消费问题的示例_python的相关文章

用Python展示动态规则法用以解决重叠子问题的示例_python

动态规划是一种用来解决定义了一个状态空间的问题的算法策略.这些问题可分解为新的子问题,子问题有自己的参数.为了解决它们,我们必须搜索这个状态空间并且在每一步作决策时进行求值.得益于这类问题会有大量相同的状态的这个事实,这种技术不会在解决重叠的子问题上浪费时间. 正如我们看到的,它也会导致大量地使用递归,这通常会很有趣. 为了说明这种算法策略,我会用一个很好玩的问题来作为例子,这个问题是我最近参加的 一个编程竞赛中的 Tuenti Challenge #4 中的第 14 个挑战问题.Train E

python使用ctypes模块调用windowsapi获取系统版本示例_python

python使用ctypes模块调用windows api GetVersionEx获取当前系统版本,没有使用python32 复制代码 代码如下: #!c:/python27/python.exe#-*- coding:utf-8 -*- "通过调用Window API判断当前系统版本"# 演示通过ctypes调用windows api函数.# 作者已经知道python32能够实现相同功能# 语句末尾加分号,纯属个人习惯# 仅作部分版本判断,更详细的版本判断推荐系统OSVERSION

Python对小数进行除法运算的正确方法示例_python

求一个算式 复制代码 代码如下: a=1 b=2 c=3   print c*(a/b) 运行结果总是0,反复检查拆开以后,发现在Python里,整数初整数,只能得出整数. 也就是 a 除 b 这个结果永远是0,只要把a或者b其中一个数改成浮点数即可. 复制代码 代码如下: a=1 b=2 c=3   print c*(a/float(b)) print c*(float(a)/b) 这样才能准确算出a除b的正确结果,当然,如果a比b大,并且不需要小数位数部分可以不用float. 如: 复制代码

windows系统中python使用rar命令压缩多个文件夹示例_python

复制代码 代码如下: #!/usr/bin/env python# Filename: backup_ver1.py import osimport time # 1. The files and directories to be backed up are specified in a list.#source=['/home/swaroop/byte','/home/swaroop/bin']source=['D:\\FileCopier\\*.*','D:\\jeecms_doc\\*.

python通过scapy获取局域网所有主机mac地址示例_python

python通过scapy获取局域网所有主机mac地址 复制代码 代码如下: #!/usr/bin/env python# -*- coding: utf-8 -*-from scapy.all import srp,Ether,ARP,confipscan='192.168.1.1/24'try:    ans,unans = srp(Ether(dst="FF:FF:FF:FF:FF:FF")/ARP(pdst=ipscan),timeout=2,verbose=False)exc

python中的内置函数getattr()介绍及示例_python

在python的官方文档中:getattr()的解释如下: getattr(object, name[, default]) Return the value of the named attribute of object. name must be a string. If the string is the name of one of the object's attributes, the result is the value of that attribute. For examp

使用Python的Scrapy框架编写web爬虫的简单示例_python

 在这个教材中,我们假定你已经安装了Scrapy.假如你没有安装,你可以参考这个安装指南. 我们将会用开放目录项目(dmoz)作为我们例子去抓取. 这个教材将会带你走过下面这几个方面:     创造一个新的Scrapy项目     定义您将提取的Item     编写一个蜘蛛去抓取网站并提取Items.     编写一个Item Pipeline用来存储提出出来的Items Scrapy由Python写成.假如你刚刚接触Python这门语言,你可能想要了解这门语言起,怎么最好的利用这门语言.假如

Python用UUID库生成唯一ID的方法示例_python

UUID介绍 UUID是128位的全局唯一标识符,通常由32字节的字符串表示.它可以保证时间和空间的唯一性,也称为GUID,全称为:UUID -- Universally Unique IDentifier,Python 中叫 UUID. 它通过MAC地址.时间戳.命名空间.随机数.伪随机数来保证生成ID的唯一性. UUID主要有五个算法,也就是五种方法来实现. uuid1()--基于时间戳.由MAC地址.当前时间戳.随机数生成.可以保证全球范围内的唯一性,但MAC的使用同时带来安全性问题,局域

在Python的Bottle框架中使用微信API的示例_python

微信这个东西估计宅男没几个不熟悉的吧,微信经过这么两年多的发展终于向开放平台跨出了友好的一步.蛋疼的以为微信会出一个详细的api等接口,兴奋不已的去申请了微信公共平台,然后开始找各种api的位置-- 花费了近一个小时,依然没找到-- 最后动用Google大杀器,终于找到了这么个链接.我了个去的,没比这还简单的api文档了吧. 最让人无法理解的是:居然没有本地开发环境支持,每次都要放在生产环境去调试. 最让人欣慰的是:就那么俩方法,生产环境调试几次也就完事了. Python(bottle)版代码如