Python多线程3：queue

Posted 2020-09-23

queue模块实现了多生产者。多消费者队列。在多线程环境下，该队列能实现多个线程间安全的信息交换。

queue模块介绍

模块实现了3种类型的队列，差别在于队列中条目检索的顺序不同。

在FIFO队列中。依照先进先出的顺序检索条目。在LIFO队列中，最后加入的条目最先检索到（操作类似一个栈）。在优先级队列中，条目被保存为有序的（使用heapq模块）而且最小值的条目被最先检索。

queue模块定义了以下的类和异常：

class queue.Queue(maxsize=0)

FIFO队列的构造器。maxsize为一个整数。表示队列的最大条目数。一旦队列满，插入将被堵塞直到队列中存在空暇空间。假设maxsize小于等于0。队列大小为无限。

class queue.LifoQueue(maxsize=0)

LIFO队列的构造器。

maxsize是一个整数，表示队列的最大条目数。一旦队列满，插入将被堵塞直到队列中存在空暇空间。假设maxsize小于等于0，队列大小为无限。

class queue.PriorityQueue(maxsize=0)

优先级队列的构造器。maxsize是一个整数，表示队列的最大条目数。一旦队列满，插入将被堵塞直到队列中存在空暇空间。

假设maxsize小于等于0，队列大小为无限。

最小值的条目被最先检索到（最小值的条目即为被sorted(list(entries))[0]返回的条目）。

通常一个条目被保存为以下的形式：(priority_number, data)。

exception queue.Empty

当Queue为空时，非堵塞的get()或者get_nowait()被调用时，将抛出该异常。

exception queue.Full

当队列满时，非堵塞的put()或者put_nowait()被调用，将抛出该异常。

Queue对象

Queue对象（Queue、LifoQueue或者PriorityQueue）提供了下面方法：


Queue.qsize()
返回队列的近似大小。注意，qsize() > 0并不能保证接下来的get()方法不被堵塞。相同。qsize() < maxsize也不能保证put()将不被堵塞。

Queue.empty()
假设队列是空的。则返回True，否则False。假设empty()返回True，并不能保证接下来的put()调用将不被堵塞。类似的。empty()返回False也不能保证接下来的get()调用将不被堵塞。

Queue.full()
假设队列满则返回True，否则返回False。假设full()返回True，并不能保证接下来的get()调用将不被堵塞。

类似的，full()返回False也不能保证接下来的put()调用将不被堵塞。

Queue.put(item, block=True, timeout=None)
放item到队列中。假设block是True，且timeout是None。该方法将一直等待直到有队列有空余空间。假设timeout是一个正整数，该方法则最多堵塞timeout秒并抛出Full异常。假设block是False而且队列满，则直接抛出Full异常（这时timeout将被忽略）。

Queue.put_nowait(item)
等价于put(item, False)。

Queue.get(block=True, timeout=None)
从队列中移除被返回一个条目。假设block是True而且timeout是None，该方法将堵塞直到队列中有条目可用。假设timeout是正整数，该方法将最多堵塞timeout秒并抛出Empty异常。假设block是False而且队列为空。则直接抛出Empty异常（这时timeout将被忽略）。

Queue.get_nowait()
等价于get(False)。

假设须要跟踪进入队列中的任务是否已经被精灵消费者线程处理完毕，能够使用以下提供的两个方法：


Queue.task_done()
表示一个先前的队列中的任务完毕了。被队列消费者线程使用。对于每一个get()获取到的任务，接下来的task_done()的调用告诉队列该任务的处理已经完毕。

假设join()调用正在堵塞，当队列中全部的条目被处理后它将恢复运行（意味着task_done()调用将被放入队列中的每一个条目接收到）。

假设调用次数超过了队列中放置的条目数目，将抛出ValueError异常。

Queue.join()
堵塞直到队列中全部条目都被获取并处理。
当一个条目被添加到队列时，未完毕任务的计数将添加。当一个消费者线程调用task_done()时，未完毕任务的计数将降低。

当未完毕任务的计数降低到0时，join()解锁。

以下是一个详细的样例。用于说明怎么等待队列任务完毕：

def worker():
    while True:
        item = q.get()
        do_work(item)
        q.task_done()

q = Queue()
for i in range(num_worker_threads):
     t = Thread(target=worker)
     t.daemon = True
     t.start()

for item in source():
    q.put(item)

q.join()       # 堵塞直到全部任务完毕