搜尋
首頁後端開發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的源码吧。
陳述
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn
详细讲解Python之Seaborn(数据可视化)详细讲解Python之Seaborn(数据可视化)Apr 21, 2022 pm 06:08 PM

本篇文章给大家带来了关于Python的相关知识,其中主要介绍了关于Seaborn的相关问题,包括了数据可视化处理的散点图、折线图、条形图等等内容,下面一起来看一下,希望对大家有帮助。

详细了解Python进程池与进程锁详细了解Python进程池与进程锁May 10, 2022 pm 06:11 PM

本篇文章给大家带来了关于Python的相关知识,其中主要介绍了关于进程池与进程锁的相关问题,包括进程池的创建模块,进程池函数等等内容,下面一起来看一下,希望对大家有帮助。

Python自动化实践之筛选简历Python自动化实践之筛选简历Jun 07, 2022 pm 06:59 PM

本篇文章给大家带来了关于Python的相关知识,其中主要介绍了关于简历筛选的相关问题,包括了定义 ReadDoc 类用以读取 word 文件以及定义 search_word 函数用以筛选的相关内容,下面一起来看一下,希望对大家有帮助。

归纳总结Python标准库归纳总结Python标准库May 03, 2022 am 09:00 AM

本篇文章给大家带来了关于Python的相关知识,其中主要介绍了关于标准库总结的相关问题,下面一起来看一下,希望对大家有帮助。

分享10款高效的VSCode插件,总有一款能够惊艳到你!!分享10款高效的VSCode插件,总有一款能够惊艳到你!!Mar 09, 2021 am 10:15 AM

VS Code的确是一款非常热门、有强大用户基础的一款开发工具。本文给大家介绍一下10款高效、好用的插件,能够让原本单薄的VS Code如虎添翼,开发效率顿时提升到一个新的阶段。

python中文是什么意思python中文是什么意思Jun 24, 2019 pm 02:22 PM

pythn的中文意思是巨蟒、蟒蛇。1989年圣诞节期间,Guido van Rossum在家闲的没事干,为了跟朋友庆祝圣诞节,决定发明一种全新的脚本语言。他很喜欢一个肥皂剧叫Monty Python,所以便把这门语言叫做python。

Python数据类型详解之字符串、数字Python数据类型详解之字符串、数字Apr 27, 2022 pm 07:27 PM

本篇文章给大家带来了关于Python的相关知识,其中主要介绍了关于数据类型之字符串、数字的相关问题,下面一起来看一下,希望对大家有帮助。

详细介绍python的numpy模块详细介绍python的numpy模块May 19, 2022 am 11:43 AM

本篇文章给大家带来了关于Python的相关知识,其中主要介绍了关于numpy模块的相关问题,Numpy是Numerical Python extensions的缩写,字面意思是Python数值计算扩展,下面一起来看一下,希望对大家有帮助。

See all articles

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅動的應用程序,用於創建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

AI Hentai Generator

AI Hentai Generator

免費產生 AI 無盡。

熱門文章

R.E.P.O.能量晶體解釋及其做什麼(黃色晶體)
2 週前By尊渡假赌尊渡假赌尊渡假赌
倉庫:如何復興隊友
4 週前By尊渡假赌尊渡假赌尊渡假赌
Hello Kitty Island冒險:如何獲得巨型種子
4 週前By尊渡假赌尊渡假赌尊渡假赌

熱工具

Safe Exam Browser

Safe Exam Browser

Safe Exam Browser是一個安全的瀏覽器環境,安全地進行線上考試。該軟體將任何電腦變成一個安全的工作站。它控制對任何實用工具的訪問,並防止學生使用未經授權的資源。

ZendStudio 13.5.1 Mac

ZendStudio 13.5.1 Mac

強大的PHP整合開發環境

SublimeText3 英文版

SublimeText3 英文版

推薦:為Win版本,支援程式碼提示!

禪工作室 13.0.1

禪工作室 13.0.1

強大的PHP整合開發環境

Dreamweaver CS6

Dreamweaver CS6

視覺化網頁開發工具