1. 什么是队列?
学过数据结构的人都知道,如果不知道队列,请Google(或百度)。
2. 在python中什么是多生产者,多消费模型?
简单来说,就是一边生产(多个生产者),一边消费(多个消费者)。比如,一边有m个线程生产数据,另一边有n个线程消费(使用)数据,这就是多生产者,多消费者模型。
注:消费依赖生产(没有厂家生产产品,就不会有关于这种产品的消费),在python中,如果生产者线程没有产生数据,那么消费者线程有关于消费行为的操作就应当处于阻塞状态。
3. 在python中为什么有queue这个标准模块?它有什么用?
用threading.Lock Objects或其他的锁来完成上一问的需求会很复杂,所以queue这个模块简化了这些操作。
queue
模块实现多生产者,多消费者队列。它特别适用于信息必须在多个线程间安全地交换的多线程程序中。该模块中的Queue
类实现了所有需要的锁定语义。
queue模块实现了三类队列,主要差别在于取得数据的顺序上。在FIFO(First In First Out,先进先出)队列中,最早加入的任务会被最先得到。在LIFO(Last In First Out,后进先出)队列中,最后加入的任务会被最先得到(就像栈一样)。在优先队列中,任务被保持有序(使用heapq
模块),拥有最小值的任务(优先级最高)被最先得到。
queue模块提供的方法:
1 Queue.empty() 2 """ 3 如果队列为空,返回True,否则返回False。如果empty()返回True,它不保证后续调用put()不会阻塞。类似的,如果empty()返回False也不能保证接下来的get()调用不会被阻塞。 4 """ 5 6 Queue.full() 7 """ 8 如果队列已满,则返回True,否则返回False。如果full()返回True,它不保证后续调用get()不会阻塞。类似的,如果full()返回False并不能保证接下来的put()调用不会被阻塞。 9 """ 10 11 Queue.qsize() 12 """ 13 返回队列的近似大小。注意,qsize()> 0不保证随后的get()不会阻塞,qsize() < maxsize也不会保证put()不会被阻塞。 14 """ 15 16 Queue.put(item, block=True, timeout=None) 17 """ 18 将item放入队列中。如果可选的参数block为True且timeout为None(默认的情况,阻塞调用,无超时),如有必要(比如队列满),阻塞调用线程,直到有空闲槽可用。如果超时是正数,则它最多阻塞超时秒,如果在该时间内没有空闲插槽,则引发Full异常。如果block为False,如果有空闲槽可用将数据放入队列,否则立即抛出Full异常(非阻塞调用,timeout被忽略)。 19 """ 20 21 Queue.get(block=True, timeout=None) 22 """ 23 从队列中移除并返回一个数据。如果可选的参数block为True且timeout为None(默认的情况,阻塞调用,无超时),阻塞调用进程直到有数据可用。如果超时是正数,则它最多阻塞超时秒,如果在该时间内没有可用的项,则引发Empty异常。如果block为False,如果有数据可用返回数据,否则立即抛出Empty异常(非阻塞调用,timeout被忽略)。 24 """ 25 26 Queue.put_nowait(item) 27 """ 28 等同于put(item, block=False)(非阻塞调用)。 29 """ 30 31 Queue.get_nowait() 32 """ 33 等同于get(block=False)(非阻塞调用)。 34 """ 35 36 Queue.task_done() 37 """ 38 意味着之前入队的一个任务已经完成。由队列的消费者线程调用。每一个get()调用得到一个任务,接下来的task_done()调用告诉队列该任务已经处理完毕。 39 40 If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue). 41 42 如果该方法被调用的次数多于被放入队列中的任务的个数,ValueError异常会被抛出。 43 """ 44 45 Queue.join() 46 """ 47 阻塞调用线程,直到队列中的所有任务被处理掉。 48 49 只要有数据被加入队列,未完成的任务数就会增加。当消费者线程调用task_done()以指示该项目已检索并且其上的所有工作都已完成时,计数将减少。当未完成的任务数降到0,join()解除阻塞。 50 """
注:由于同一进程的线程之间内存空间是共享的,故在同一进程的任一线程中定义的Queue对象,在该进程的任一线程都可以使用。
例(演示queue模块的使用):
1 import queue,threading 2 import random,time 3 4 class Producter(threading.Thread): 5 def __init__(self, name): 6 super(Producter, self).__init__() 7 self.name = "Producter:" + str(name) 8 9 def run(self): 10 things = ['A', 'B', 'C'] 11 for i in range(3): 12 production = random.choice(things) 13 print(self.name + " producted--->" + production) 14 production = production + " from " + self.name 15 all_productions.put(production) #将生产的数据放入队列 16 time.sleep(1) 17 time.sleep(5) 18 19 class Consumer(threading.Thread): 20 def __init__(self, name): 21 super(Consumer, self).__init__() 22 self.name = "Consumer:" + str(name) 23 24 def run(self): 25 for i in range(2): 26 thing = all_productions.get() #拿出已经生产好的数据 27 print(self.name + " is using--->" + thing) 28 time.sleep(2) 29 all_productions.task_done() #告诉队列有关这个数据的任务已经完成 30 time.sleep(5) 31 32 if __name__ == "__main__": 33 all_productions = queue.Queue() 34 35 #启动两个生产者线程生 36 p_s = [] 37 for i in range(2): 38 p = Producter(i) 39 p.start() 40 p_s.append(p) 41 42 #启动三个消费者线程消费 43 c_s = [] 44 for i in range(3): 45 c = Consumer(i) 46 c.start() 47 c_s.append(c) 48 49 #阻塞,直到生产者生产的数据被消耗完 50 all_productions.join() 51 #等待生产者线程结束 52 for p in p_s: 53 p.join() 54 #等待消费者线程结束 55 for c in c_s: 56 c.join() 57 print("finish....")