queue 介绍
queue 是 python 中的标准库，俗称队列，可以直接 import 引用，在 python2.x 中，模块名为 Queue。
Python2.x 是 import Queue 注意Q是大写。 Python3.x 变成了queue。
在 python 中，多个线程之间的数据是共享的，多个线程进行数据交换的时候，不能够保证数据的安全性和一致性，所以当多个线程需要进行数据交换的时候，队列就出现了，队列可以完美解决线程间的数据交换，保证线程间数据的安全性和一致性
queue 是多线程中的使用的栈，但是Python 解释器有一个全局解释器锁(PIL)，导致每个 Python 进程中最多同时运行一个线程，因此 Python 多线程程序并不能改善程序性能，不能发挥多核系统的优势。
multiprocessing.Queue 是Python 2.6 引入的用来实现多进程的一种高性能栈。
collections.deque 是为了高效实现插入和删除操作的双向列表，适合用于队列和栈。
Python 虽然不能利用 多线程 实现多核任务，但可以通过 多进程 实现多核任务。
Python 队列 模块 queue 多应用在多线程应用中，多线程访问共享变量。对于多线程而言，访问共享变量时，python 队列 queue 是线程安全的。从queue 队列的具体实现中，可以看出 queue 使用了1个 线程互斥锁(pthread.Lock())，以及3个条件标量 (pthread.condition()) ，来保证了线程安全。即 python 的 queue 设计的是 线程安全 的。内部实现是在抢占式线程加上临时锁。但是没有涉及如何去处理线程的重入。
Python 的 queue 模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列PriorityQueue。这些队列都实现了锁原语，能够在多线程中直接使用。可以使用队列来实现线程间的同步。
queue队列的互斥锁和条件变量 - python线程中同步锁：https://blog.csdn.net/hehe123456zxc/article/details/52264829
Python 中的 Queue 实现了一个同步队列，并且该类实现了所有需要的锁原语。Queue实现了三种队列：普通的FIFO队列（Queue）、LIFO队列（LifoQueue）、优先级队列（PriorityQueue）。其使用方法类似。
下面以普通的 先进先出队列 Queue 为例谈一下 Queue 中的主要方法:
#引入Queue类 from Queue import Queue #得到队列的大小 Queue.qsize() #判断队列是否为空 Queue.empty() #判断队列是否已满 Queue.full() #从队列头获取元素，默认为阻塞 Queue.get([block[,timeout]]) #从队列头获取元素，非阻塞方式 Queue.get_nowait() #或者 Queue.get(block=False) #阻塞写入队列 Queue.put(item) #非阻塞写入队列 Queue.put_nowait(item) #或者 Queue.put(item,block=False) #向队列中已完成的元素发送join信号 Queue.task_done()
上面从队列中获取元素和向队列中添加元素都有阻塞和非阻塞的方式，采用阻塞方式，如果从队列中取元素而元素为空，则线程会停下来等待知道队列中有元素可以取出；如果向队列中添加元素而此时队列已满，则同样线程会停下来直到停止。如果采用非阻塞方式，取元素时一旦队列为空，则会引发Empty异常，放元素时一旦队列已满，就会引发Full异常。
下面是采用 Queue 实现的经典生产者消费者问题的代码：http://blog.itpub.net/22664653/viewspace-764044/
#!/usr/bin/python3 # -*- coding: utf-8 -*- # @Author : # @File : text.py # @Software : PyCharm # @description : XXX from queue import Queue import random import threading import time # Producer thread class Producer(threading.Thread): def __init__(self, t_name, queue): threading.Thread.__init__(self, name=t_name) self.data = queue def run(self): for i in range(5): print("%s: %s is producing %d to the queue!" % (time.ctime(), self.getName(), i)) self.data.put(i) time.sleep(random.randrange(10) / 5) print("%s: %s finished!" % (time.ctime(), self.getName())) # Consumer thread class Consumer(threading.Thread): def __init__(self, t_name, queue): threading.Thread.__init__(self, name=t_name) self.data = queue def run(self): for i in range(5): val = self.data.get() print("%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(), self.getName(), val)) time.sleep(random.randrange(10)) print("%s: %s finished!" % (time.ctime(), self.getName())) # Main thread def main(): queue = Queue() producer = Producer('Pro.', queue) consumer = Consumer('Con.', queue) producer.start() consumer.start() producer.join() consumer.join() print('All threads terminate!') if __name__ == '__main__': main()
python 中有三种 Queue
import Queue
myqueue = Queue.Queue() # python 自带库 的 队列
from multiprocessing import Queue # multiprocessing 模块中队列
myqueue = Queue.Queue()
import multiprocessing
manager = multiprocessing.Manager() # multiprocessing 模块中 manager 的队列
myqueue=manager.Queue()
Queue.Queue 是进程内非阻塞队列，且各自进程私有。multiprocess.Queue 是跨进程通信队列，各子进程共有。
Manager 是 multiprocessing 的封装 .Manager.Queue 和 Queue, multiprocessing.Queue 没有太大关系。
测试 示例 2（进程间 通讯：生产者 - 消费者）：
import random import time from multiprocessing import Queue from multiprocessing import Process q_1 = Queue() q_2 = Queue() def run(): q_1.put(3) # print(q.get()) def consumer(share_q): while True: t = share_q.get() if t: print('consumer : {0}'.format(t)) else: time.sleep(0.5) def producer(share_q): while True: t = random.randint(1, 100) share_q.put(t) print('producer : {0}'.format(t)) time.sleep(1) pass def test_1(): t = Process(target=run) t.start() # time.sleep(1) print(q_1.get()) def test_2(): p_producer = Process(target=producer, args=(q_2,)) p_consumer = Process(target=consumer, args=(q_2,)) p_producer.start() p_consumer.start() pass if __name__ == '__main__': # test_1() test_2()
Python 自带库 中 queue 模块有三种队列及构造函数:
Python queue 模块的 FIFO 队列 先进先出。 class queue.Queue(maxsize)
LIFO 类似于堆，即先进后出。 class queue.LifoQueue(maxsize)
还有一种是优先级队列，级别越低越先出来。 class queue.PriorityQueue(maxsize)
queue.Queue(maxsize=0)：构造一个FIFO(先进先出)的队列
queue.LifoQueue(maxsize=0)：构造一个LIFO(后进先出)的队列
queue.PriorityQueue(maxsize=0)：构造一个具有优先级的队列,存储的是一个元组(n, value),n为数字代表优先级,数字越小,级别越高
queue 模块中的常用方法:
queue.qsize() ： 返回队列的大小
queue.empty() ： 如果队列为空，返回True，反之False
queue.full() ： 如果队列满了，返回True，反之False
queue.full ： 与 maxsize 大小对应
queue.get([block[, timeout]]) ： 获取队列，timeout等待时间。从队列中获取任务，并且从队列中移除此任务。首先尝试获取互斥锁，获取成功则队列中get任务，如果此时队列为空，则wait等待生产者线程添加数据。get到任务后，会调用self.not_full.notify()通知生产者线程，队列可以添加元素了。最后释放互斥锁
queue.get_nowait() ： 相当queue.get(False)。无阻塞的向队列中get任务，当队列为空时，不等待，而是直接抛出empty异常，重点是理解block=False
queue.put(item) ： 写入队列，timeout 等待时间。申请获得互斥锁，获得后，如果队列未满，则向队列中添加数据，并通知notify其它阻塞的某个线程，唤醒等待获取require互斥锁。如果队列已满，则会wait等待。最后处理完成后释放互斥锁。其中还有阻塞block以及非阻塞，超时等逻辑
queue.put_nowait(item) ： 相当queue.put(item, False)。无阻塞的向队列中添加任务，当队列为满时，不等待，而是直接抛出full异常，重点是理解block=False
queue.task_done() 在完成一项工作之后，queue.task_done()函数向任务已经完成的队列发送一个信号
queue.join() 实际上意味着等到队列为空，再执行别的操作。待队列中任务全部处理完毕，需要配合queue.task_done使用
这个模块定义了两个异常
queue.Empty：如果队列中为空,继续调用非阻塞的get_nowait()会抛出异常
queue.Full：如果队列已满,继续调用非阻塞的put_nowait()会抛出异常
类 和 异常
import Queue #类 Queue.Queue(maxsize = 0) #构造一个FIFO队列,maxsize设置队列大小的上界, 如果插入数据时, 达到上界会发生阻塞, 直到队列可以放入数据. 当maxsize小于或者等于0, 表示不限制队列的大小(默认) Queue.LifoQueue(maxsize = 0) #构造一LIFO队列,maxsize设置队列大小的上界, 如果插入数据时, 达到上界会发生阻塞, 直到队列可以放入数据. 当maxsize小于或者等于0, 表示不限制队列的大小(默认) Queue.PriorityQueue(maxsize = 0) #构造一个优先级队列,,maxsize设置队列大小的上界, 如果插入数据时, 达到上界会发生阻塞, 直到队列可以放入数据. 当maxsize小于或者等于0, 表示不限制队列的大小(默认). 优先级队列中, 最小值被最先取出 #异常 Queue.Empty #当调用非阻塞的get()获取空队列的元素时, 引发异常 Queue.Full #当调用非阻塞的put()向满队列中添加元素时, 引发异常
队列对象
三种队列对象提供的公共的方法
Queue.empty() #如果队列为空, 返回True(注意队列为空时, 并不能保证调用put()不会阻塞); 队列不空返回False(不空时, 不能保证调用get()不会阻塞) Queue.full() #如果队列为满, 返回True(不能保证调用get()不会阻塞), 如果队列不满, 返回False(并不能保证调用put()不会阻塞) Queue.put(item[, block[, timeout]]) #向队列中放入元素, 如果可选参数block为True并且timeout参数为None(默认), 为阻塞型put(). 如果timeout是正数, 会阻塞timeout时间并引发Queue.Full异常. 如果block为False为非阻塞put Queue.put_nowait(item) #等价于put(itme, False) Queue.get([block[, timeout]]) #移除列队元素并将元素返回, block = True为阻塞函数, block = False为非阻塞函数. 可能返回Queue.Empty异常 Queue.get_nowait() #等价于get(False) Queue.task_done() #在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号 Queue.join() #实际上意味着等到队列为空，再执行别的操作
queue 使用示例
下面是官方文档给多出的多线程模型:
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() # block until all tasks are done
改写的官方文档代码 ：
控制线程退出 可以参考：http://bbs.chinaunix.net/forum.php?mod=viewthread&tid=4131893
import queue import threading # 要开启 的 线程 数 num_worker_threads = 3 source = [100, 200, 300, 400, 500, 600, 700, 800, 900] def do_work(*args): info = '[ thread id {0}]:{1}'.format(args[0], args[1]) print(info) def worker(t_id): while True: item = q.get() if item is None: break do_work(t_id, item) q.task_done() q = queue.Queue() threads = [] for index in range(num_worker_threads): t = threading.Thread(target=worker, args=(index,)) t.start() threads.append(t) for item in source: q.put(item) # block until all tasks are done q.join() # stop workers for i in range(num_worker_threads): q.put(None) for t in threads: t.join()
基本使用示例：
import queue # 以下三个队列都可以设置最大长度maxsize,默认是无限大 print("-------------queue.Queue----------------") # 线程消息队列,FIFO(先进先出) q = queue.Queue() q.put("one") q.put("two") q.put("three") print(q.get()) print(q.get()) print(q.get()) try: # 队列中没有数据, 会阻塞。 # 阻塞时间到了还没有数据 抛出 queue.Empty 异常 print(q.get(timeout=3)) except queue.Empty as q_e: print('queue empty') print("-------------queue.LifoQueue----------------") # 线程消息队列,LIFO(后进先出) lq = queue.LifoQueue() lq.put("one") lq.put("two") lq.put("three") print(lq.get()) print(lq.get()) print(lq.get()) print("-------------queue.PriorityQueue----------------") # 线程消息队列,PriorityQueue(优先级的队列:数字越小优先级越高) pq = queue.PriorityQueue() pq.put((1, "Jet")) pq.put((3, "Jack")) pq.put((2, "Judy")) print(pq.get()) print(pq.get()) print(pq.get())
执行结果： -------------queue.Queue---------------- one two three queue empty -------------queue.LifoQueue---------------- three two one -------------queue.PriorityQueue---------------- (1, 'Jet') (2, 'Judy') (3, 'Jack')
Python2.7 Queue模块学习（生产者 - 消费者）：https://my.oschina.net/guol/blog/93275
#!/usr/bin/python3 # -*- coding: utf-8 -*- # @Author : # @File : text.py # @Software : PyCharm # @description : XXX import queue import threading Thread_id = 1 Thread_num = 3 class MyThread(threading.Thread): def __init__(self, q): global Thread_id super(MyThread, self).__init__() self.q = q self.Thread_id = Thread_id Thread_id = Thread_id + 1 def run(self): while True: try: # 不设置阻塞的话会一直去尝试获取资源 task = self.q.get(block=True, timeout=1) except queue.Empty as e: info_e = 'Thread ' + str(self.Thread_id) + ' end' print(info_e) break # 取到数据，开始处理（依据需求加处理代码） info_d = "Starting " + str(self.Thread_id) print(info_d) print(task) self.q.task_done() info_end = "Ending " + str(self.Thread_id) print(info_end) q_test = queue.Queue(10) # 向资源池里面放10个数用作测试 for i in range(10): q_test.put(i) # 开Thread_num个线程 for i in range(0, Thread_num): worker = MyThread(q_test) worker.start() # 等待所有的队列资源都用完 q_test.join() print("Exiting Main Thread")
q.task_done是表明当前的资源处理完了，q.join()会等到所有的资源都被处理了才会向下继续执行，这就是一种同步。
multiprocessing.Queue
用于多进程，multiprocess.Queue 是跨进程通信队列。Manager 是 multiprocessing 的封装 ，Manager.Queue 和 Queue, multiprocessing.Queue 没有太大关系。
multiprocessing supports two types of communication channel between processes:
multiprocessing 支持两种类型的进程间通信方式 queues 和 pips。
multiprocessing.Queue 使用示例（此程序是在队列中加入10个数字，然后用2个进程来取出）：
import multiprocessing import time if __name__ == '__main__': # 创建消息队列 # 3： 表示消息队列最大个数 queue = multiprocessing.Queue(3) # 放入数据 queue.put(1) queue.put("abc") queue.put(["abc", "456"]) # 队列满了在放入数据, 就不能放入数据了，直到消息队列有空闲位置才能再放入数据 # queue.put(("34", 90)) # put_nowait： 不会等待队列有空闲位置再放入数据，如果数据放入不成功就直接崩溃 # queue.put_nowait(("34", 90)) # 建议: 放入数据使用put，因为比较安全不会崩溃 # 查看队列是否是满的 # result = queue.full() # print(result) # 坑点: 只使用put放入数据直接判断队列是否为空获取的结果不正确，因为没有等队列把数据写完直接就取获取了，那么这是队列是空的 # for i in range(10000): # print(i) # 解决办法: 1. 延时一段时间 2. 通过个数去判断 # time.sleep(0.001) # if queue.qsize() == 0: # mac 版本不能使用qsize() print("队列是空的") # result = queue.empty() # print(result) # 查看队列的个数 size = queue.qsize() print("消息队列个数:", size) # 获取队列的数据 value = queue.get() print(value) size = queue.qsize() print("消息队列个数:", size) # 获取队列的数据 value = queue.get() print(value) size = queue.qsize() print("消息队列个数:", size) # 获取队列的数据 value = queue.get() print(value) size = queue.qsize() print("消息队列个数:", size) # 队列为空， 使用get会等待，直到队列有数据以后再取值 # value = queue.get() # print(value) # 队列为空，取值的时候不等待，但是取不到值那么直接崩溃了 # value = queue.get_nowait() # print(value) # 建议： 获取队列的数据统一get，因为能保证代码不会有问题
#!/usr/bin/env python3 import time from multiprocessing import Process, Queue def func_a(share_q): while True: try: num = share_q.get_nowait() print('我是进程A,取出数字:%d' % num) time.sleep(1) except BaseException as e: break def func_b(share_q): while True: try: num = share_q.get_nowait() print('我是进程B,取出数字:%d' % num) time.sleep(1) except BaseException as e: break if __name__ == '__main__': q = Queue() # 创建列队，不传数字表示列队不限数量 for i in range(11): q.put(i) p1 = Process(target=func_a, args=(q,)) p2 = Process(target=func_b, args=(q,)) p1.start() p2.start()
使用进程池 Pool时，Queue 会出错，需要使用 Manager.Queue：
#!/usr/bin/env python3 import time from multiprocessing import Pool, Manager, Queue def func_a(p_id, share_q): num = share_q.get_nowait() print('我是进程%d, 取出数字:%d' % (p_id, num)) time.sleep(1) if __name__ == '__main__': q = Manager().Queue() for i in range(11): q.put(i) pool = Pool(3) for i in range(10): pool.apply_async(func_a, (i, q)) pool.close() pool.join()
主进程定义了一个 Queue 类型的变量，并作为 Process 的 args 参数传给子进程 processA 和 processB
两个进程一个向队列中写数据，一个读数据。
import time from multiprocessing import Process, Queue MSG_QUEUE = Queue(5) def start_a(msgQueue): while True: if msgQueue.empty() > 0: print('queue is empty %d' % (msgQueue.qsize())) else: msg = msgQueue.get() print('get msg %s' % (msg,)) time.sleep(1) def start_b(msgQueue): while True: msgQueue.put('hello world') print('put hello world queue size is %d' % (msgQueue.qsize(),)) time.sleep(3) if __name__ == '__main__': processA = Process(target=start_a, args=(MSG_QUEUE,)) processB = Process(target=start_b, args=(MSG_QUEUE,)) processA.start() print('processA start..') processB.start() print('processB start..')
collections.deque
先来看官方文档：
有人对比过以上三者的性能，deque 作为一种双向队列性能完胜其他两者。
from collections import deque d = deque('ghi') # 使用 'ghi' 创建一个具有3个元素的队列 for elem in d: # 迭代 队列中的元素 print(elem.upper()) d.append('j') # 在队列 右边 添加一个元素 d.appendleft('f') # 在队列 左边 添加一个元素 print(d) d = deque(['f', 'g', 'h', 'i', 'j']) print(d.pop()) # return and remove the rightmost item print(d.popleft()) # return and remove the leftmost item print(list(d)) # list the contents of the deque print(d[0]) # peek at leftmost item print(d[-1]) # peek at rightmost item print(list(reversed(d))) # list the contents of a deque in reverse print('h' in d) # search the deque d.extend('jkl') # add multiple elements at once print(d) d = deque(['g', 'h', 'i', 'j', 'k', 'l']) d.rotate(1) # right rotation print(d) d = deque(['l', 'g', 'h', 'i', 'j', 'k']) d.rotate(-1) # left rotation print(d) d = deque(['g', 'h', 'i', 'j', 'k', 'l']) print(deque(reversed(d))) # make a new deque in reverse order d = deque(['l', 'k', 'j', 'i', 'h', 'g']) d.clear() # empty the deque # print(d.pop()) # cannot pop from an empty deque d.extendleft('abc') # extendleft() reverses the input order print(d)
最后附上Queue模块的源码
直接点进去queue.py，源码只有249行，还好，看下源码结构
点开两个异常，非常简单，继承Exception而已，我们更关注__all__
1) all
all：在模块级别暴露公共接口，比如在导库的时候不建议写 from xxx import *，因为会把xxx模块里所有非下划线开头的成员都
引入到当前命名空间中，可能会污染当前命名空间。如果显式声明了 all*，import 就只会导入 all 列出的成员。
(不建议使用：from xxx import *** 这种语法！！！)
接着看下Queue类结构，老规矩，先撸下init方法
文档注释里写了：创建一个maxsize大小的队列，如果<=0，队列大小是无穷的。设置了maxsize，然后调用self._init(maxsize)，点进去看下：
这个deque是什么？
2) deque类
其实是collections模块提供的双端队列，可以从队列头部快速增加和取出对象，对应两个方法：popleft()与appendleft()，时间复杂度只有O(1)，相比起list对象的insert(0,v)和pop(0)的时间复杂度为O(N)，列表元素越多，元素进出耗时会越长！
回到源码，接着还定义了：
mutex：threading.Lock()，定义一个互斥锁
not_empty = threading.Condition(self.mutex)：定义一个非空的条件变量
not_full = threading.Condition(self.mutex)：定义一个非满的条件变量
all_tasks_done = threading.Condition(self.mutex)：定义一个任务都完成的条件变量
unfinished_tasks = 0：初始化未完成的任务数量为0
接着到task_done()方法：
with加锁，未完成任务数量-1，判断未完成的任务数量，小于0，抛出异常：task_done调用次数过多，等于0则唤醒所有等待线程，修改未完成任务数量；
再接着到join()方法：
with加锁，如果还有未完成的任务，wait堵塞调用者进程；
接下来是qsize，empty和full函数，with加锁返回大小而已：
接着是put()函数：
with加锁，判断maxsize是否大于0，上面也讲了maxsize<=0代表队列是可以无限扩展的，那就不存在队列满的情况，maxsize<=0的话直接就往队列里放元素就可以了，同时未完成任务数+1，随机唤醒等待线程。
如果maxsize大于0代表有固定容量，就会出现队列满的情况，就需要进行细分了：
1.block为False：非堵塞队列，判断当前大小是否大于等于容量，是，抛出Full异常；
2.block为True，没设置超时：堵塞队列，判断当前大小是否大于等于容量，
是，堵塞线程；
3.block为True，超时时间<0：直接抛出ValueError异常，超时时间应为非负数；
4.block为True，超时时间>=0，没倒时间堵塞线程，到时间抛出Full异常；
再接着是get()函数，和put()类似，只是抛出的异常为：Empty
这两个就不用说了，非堵塞put()和get()，最后就是操作双端队列的方法而已；
另外两种类型的队列也非常简单，继承Queue类，然后重写对应的四个方法而已～
3) heapq模块
PriorityQueue优先级队里的heappush()和heappop()是heapq模块提供的两个方法，heap队列，q队列，堆一般可看做是一棵树的数组对象(二叉树堆)，规则如下：某个节点的值总是不大于或不小于其孩子节点的值
然后又分最大堆和最小堆：
利用：heappush()可以把数据放到堆里，会自动按照二叉树的结构进行存储；
利用：heappop(heap)：从heap堆中删除最小元素，并返回，heap再按完全二叉树规范重排；
queue.py模块大概的流程就是这个样子咯，总结下套路把：
关键点核心：三个条件变量，
not_empty：get的时候，队列空或在超时时间内，堵塞读取线程，非空唤醒读取线程；
not_full：put的时候，队列满或在超时时间内，堵塞写入线程，非满唤醒写入线程；
all_tasks_done：未完成任务unfinished_tasks不为0的时候堵塞调用队列的线程，
未完成任务不为0时唤醒所有调用队列的线程；
源码：
"""A multi-producer, multi-consumer queue.""" from time import time as _time try: import threading as _threading # 导入threading模块 except ImportError: import dummy_threading as _threading # 该模块的接口和thread相同，在没有实现thread模块的平台上提供thread模块的功能。 from collections import deque # https://github.com/BeginMan/pythonStdlib/blob/master/collections.md import heapq # 堆排序 https://github.com/qiwsir/algorithm/blob/master/heapq.md __all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue'] # 模块级别暴露接口 class Empty(Exception): """当调用Queue.get(block=0)/get_nowait()时触发Empty异常 调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。 如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。 如果队列为空且block为False，队列将引发Empty异常 """ pass class Full(Exception): """当调用Queue.put(block=0)/put_nowait()时触发Full异常 如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。 如果block为0，put方法将引发Full异常。 """ pass class Queue: """创建一个给定的最大大小的队列对象. FIFO(先进先出)队列, 第一加入队列的任务, 被第一个取出 If maxsize is <= 0, the queue size is 无限大小. """ def __init__(self, maxsize=0): self.maxsize = maxsize self._init(maxsize) # 初始化queue为空 # 所有获取锁的方法必须在返回之前先释放，互斥锁在下面三个Condition条件共享 # 从而获取和释放的条件下也获得和释放互斥锁。 self.mutex = _threading.Lock() # Lock锁 # 当添加queue元素时通知`not_empty`,之后线程等待get self.not_empty = _threading.Condition(self.mutex) # not_empty Condition实例 # 当移除queue元素时通知`not_full`,之后线程等待put. self.not_full = _threading.Condition(self.mutex) # not_full Condition实例 # 当未完成的任务数为0时，通知`all_tasks_done`,线程等待join() self.all_tasks_done = _threading.Condition(self.mutex) # all_tasks_done Condition实例 self.unfinished_tasks = 0 def task_done(self): """表明，以前排队的任务完成了 被消费者线程使用. 对于每个get()，随后调用task_done()告知queue这个task已经完成 """ self.all_tasks_done.acquire() try: # unfinished_tasks 累减 unfinished = self.unfinished_tasks - 1 if unfinished <= 0: # 调用多次task_done则触发异常 if unfinished < 0: raise ValueError('task_done() called too many times') self.all_tasks_done.notify_all() # 释放所有等待该条件的线程 self.unfinished_tasks = unfinished finally: self.all_tasks_done.release() def join(self): """阻塞直到所有任务都处理完成 未完成的task会在put()累加，在task_done()累减, 为0时，join()非阻塞. """ self.all_tasks_done.acquire() try: # 一直循环检查未完成数 while self.unfinished_tasks: self.all_tasks_done.wait() finally: self.all_tasks_done.release() def qsize(self): """返回队列的近似大小（不可靠！）""" self.mutex.acquire() n = self._qsize() # len(queue) self.mutex.release() return n def empty(self): """队列是否为空(不可靠).""" self.mutex.acquire() n = not self._qsize() self.mutex.release() return n def full(self): """队列是否已满(不可靠!).""" self.mutex.acquire() n = 0 < self.maxsize == self._qsize() self.mutex.release() return n def put(self, item, block=True, timeout=None): """添加元素. 如果可选参数block为True并且timeout参数为None(默认), 为阻塞型put(). 如果timeout是正数, 会阻塞timeout时间并引发Queue.Full异常. 如果block为False为非阻塞put """ self.not_full.acquire() try: if self.maxsize > 0: if not block: if self._qsize() == self.maxsize: raise Full elif timeout is None: while self._qsize() == self.maxsize: self.not_full.wait() elif timeout < 0: raise ValueError("'timeout' must be a non-negative number") else: endtime = _time() + timeout while self._qsize() == self.maxsize: remaining = endtime - _time() if remaining <= 0.0: raise Full self.not_full.wait(remaining) self._put(item) self.unfinished_tasks += 1 self.not_empty.notify() finally: self.not_full.release() def put_nowait(self, item): """ 非阻塞put 其实就是将put第二个参数block设为False """ return self.put(item, False) def get(self, block=True, timeout=None): """移除列队元素并将元素返回. block = True为阻塞函数, block = False为非阻塞函数. 可能返回Queue.Empty异常 """ self.not_empty.acquire() try: if not block: if not self._qsize(): raise Empty elif timeout is None: while not self._qsize(): self.not_empty.wait() elif timeout < 0: raise ValueError("'timeout' must be a non-negative number") else: endtime = _time() + timeout while not self._qsize(): remaining = endtime - _time() if remaining <= 0.0: raise Empty self.not_empty.wait(remaining) item = self._get() self.not_full.notify() return item finally: self.not_empty.release() def get_nowait(self): """ 非阻塞get() 也即是get()第二个参数为False """ return self.get(False) # Override these methods to implement other queue organizations # (e.g. stack or priority queue). # These will only be called with appropriate locks held # 初始化队列表示 def _init(self, maxsize): self.queue = deque() # 将queue初始化为一个空的deque对象 def _qsize(self, len=len): # 队列长度 return len(self.queue) # Put a new item in the queue def _put(self, item): self.queue.append(item) # Get an item from the queue def _get(self): return self.queue.popleft() class PriorityQueue(Queue): """ 继承Queue 构造一个优先级队列 maxsize设置队列大小的上界, 如果插入数据时, 达到上界会发生阻塞, 直到队列可以放入数据. 当maxsize小于或者等于0, 表示不限制队列的大小(默认). 优先级队列中, 最小值被最先取出 """ def _init(self, maxsize): self.queue = [] def _qsize(self, len=len): return len(self.queue) def _put(self, item, heappush=heapq.heappush): heappush(self.queue, item) def _get(self, heappop=heapq.heappop): return heappop(self.queue) class LifoQueue(Queue): """ 构造一LIFO(先进后出)队列 maxsize设置队列大小的上界, 如果插入数据时, 达到上界会发生阻塞, 直到队列可以放入数据. 当maxsize小于或者等于0, 表示不限制队列的大小(默认) """ def _init(self, maxsize): self.queue = [] def _qsize(self, len=len): return len(self.queue) def _put(self, item): self.queue.append(item) def _get(self): return self.queue.pop() # 与Queue相比，仅仅是 将popleft()改成了pop()