发布于 3年前

Python多线程编程 - 用队列实现生产者消费者模式

生产者和消费者模式的实现有很多中方式。这里的示例基于队列来实现生产者和消费者模式。

生产者和消费者在各自的线程中运行,并检查队列的状态。如果队列没有满,生产者线程负责往队列里放数据。消费者线程负责从队列里取数据,前提是队列不是空的。

代码示例:

import threading
import time
import logging
import random
import Queue
logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-9s) %(message)s',)
BUF_SIZE = 10
q = Queue.Queue(BUF_SIZE)
class ProducerThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, verbose=None):
        super(ProducerThread,self).__init__()
        self.target = target
        self.name = name
    def run(self):
        while True:
            if not q.full():
                item = random.randint(1,10)
                q.put(item)
                logging.debug('Putting ' + str(item)  
                              + ' : ' + str(q.qsize()) + ' items in queue')
                time.sleep(random.random())
        return
class ConsumerThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, verbose=None):
        super(ConsumerThread,self).__init__()
        self.target = target
        self.name = name
        return
    def run(self):
        while True:
            if not q.empty():
                item = q.get()
                logging.debug('Getting ' + str(item) 
                              + ' : ' + str(q.qsize()) + ' items in queue')
                time.sleep(random.random())
        return
if __name__ == '__main__':
    p = ProducerThread(name='producer')
    c = ConsumerThread(name='consumer')
    p.start()
    time.sleep(2)
    c.start()
    time.sleep(2)

输出结果:

(producer ) Putting 2 : 1 items in queue
(producer ) Putting 10 : 2 items in queue
(producer ) Putting 6 : 3 items in queue
(producer ) Putting 7 : 4 items in queue
(producer ) Putting 1 : 5 items in queue
(consumer ) Getting 2 : 4 items in queue
(consumer ) Getting 10 : 3 items in queue
(producer ) Putting 1 : 4 items in queue
(producer ) Putting 8 : 5 items in queue
(consumer ) Getting 6 : 4 items in queue
(producer ) Putting 10 : 5 items in queue
...

1、队列包含了Condition,其中Condition有锁,所以我们不需要关心队列Condition和锁。

2、生产者调用Queue.put(item[, block[, timeout]])方法往队列里插数据,在生产者插入数据前,它需要获取锁。其中参数block如果为true,而timeout默认为None,那么生产者会一直阻塞知道有释放的槽。如果timeout是一个正数,那么生产者会等到timeout 设定的秒后,如果还没有获取到锁,那么会抛出一个FullException。

3、put()函数也会检查队列是否满了,如果满的话,它会内部调用wait方法,这样生产者就开始等待。

4、消费者调用Queue.get([block[, timeout]]),在执行删除数据前,它也需要获取锁。如果队列是空的,那么消费者进入等待状态。

©2020 edoou.com   京ICP备16001874号-3