生产者-消费者问题和Queue/queue 模块

最后一个例子演示了生产者-消费者模型这个场景。在这个场景下，商品或服务的生产者
生产商品，然后将其放到类似队列的数据结构中。生产商品的时间是不确定的，同样消费者
消费生产者生产的商品的时间也是不确定的。

我们使用Queue 模块（Python 2.x 版本，在Python 3.x 版本中重命名为queue）来提供线
程间通信的机制，从而让线程之间可以互相分享数据。具体而言，就是创建一个队列，让生
产者（线程）在其中放入新的商品，而消费者（线程）消费这些商品。表4-5 列举了这个模
块中的一些属性。

我们将使用示例4-12（prodcons.py）来演示生产者-消费者Queue/queue。下面是这个脚
本某次执行的输出。

$ prodcons.py
starting writer at: Sun Jun 18 20:27:07 2006
producing object for Q... size now 1
starting reader at: Sun Jun 18 20:27:07 2006
consumed object from Q... size now 0
producing object for Q... size now 1
consumed object from Q... size now 0
producing object for Q... size now 1
producing object for Q... size now 2
producing object for Q... size now 3
consumed object from Q... size now 2
consumed object from Q... size now 1
writer finished at: Sun Jun 18 20:27:17 2006
consumed object from Q... size now 0
reader finished at: Sun Jun 18 20:27:25 2006
al l DONE

示例4-12 生产者-消费者问题（prodcons.py）

from random import randint 
from time import sleep
from queue import Queue  # 修正为 Python 3 的 queue
import threading  # 添加标准线程库
 
# 自定义线程类实现 
class MyThread(threading.Thread):
    def __init__(self, func, args, name=''):
        super().__init__()
        self.func  = func
        self.args  = args
        self.name  = name
        
    def run(self):
        self.func(*self.args) 
 
def writeQ(queue):
    print('producing object for Q...')
    queue.put('xxx')   # 移除过时的 block 参数 
    print("size now", queue.qsize()) 
 
def readQ(queue):
    val = queue.get()   # 移除过时的 block 参数
    print('consumed object from Q... size now', queue.qsize()) 
 
def writer(queue, loops):
    for _ in range(loops):
        writeQ(queue)
        sleep(randint(1, 3))
 
def reader(queue, loops):
    for _ in range(loops):
        readQ(queue)
        sleep(randint(2, 5))
 
funcs = [writer, reader]
nfuncs = range(len(funcs))
 
def main():
    nloops = randint(2, 5)
    q = Queue(32)  # 添加队列容量限制
    
    threads = []
    for i in nfuncs:
        t = MyThread(funcs[i], (q, nloops), funcs[i].__name__)
        threads.append(t) 
    
    # 修正缩进问题 
    for i in nfuncs:
        threads[i].start()
    
    for i in nfuncs:
        threads[i].join()
    
    print('all DONE')
 
if __name__ == '__main__':
    main()

运行结果

producing object for Q...
size now 1
consumed object from Q... size now 0
producing object for Q...
size now 1
consumed object from Q... size now 0
producing object for Q...
size now 1
producing object for Q...
size now 2
consumed object from Q... size now 1
producing object for Q...
size now 2
consumed object from Q... size now 1
consumed object from Q... size now 0
all DONE