Queue
创新互联自2013年创立以来，是专业互联网技术服务公司，拥有项目网站设计制作、成都网站制作网站策划，项目实施与项目整合能力。我们以让每一个梦想脱颖而出为使命，1280元巩留做网站,已为上家服务,为巩留各地企业和个人服务,联系电话:18982081108
Queue是python标准库中的线程安全的队列（FIFO）实现,提供了一个适用于多线程编程的先进先出的数据结构，即队列，用来在生产者和消费者线程之间的信息传递
基本FIFO队列
class Queue.Queue(maxsize=0)
FIFO即First in First Out,先进先出。Queue提供了一个基本的FIFO容器，使用方法很简单,maxsize是个整数，指明了队列中能存放的数据个数的上限。一旦达到上限，插入会导致阻塞，直到队列中的数据被消费掉。如果maxsize小于或者等于0，队列大小没有限制。
举个栗子：
import Queue
q = Queue.Queue()for i in range(5):
q.put(i)while not q.empty(): print q.get()
输出：
1
2
3
4
LIFO队列
class Queue.LifoQueue(maxsize=0)
LIFO即Last in First Out,后进先出。与栈的类似，使用也很简单,maxsize用法同上
再举个栗子：
import Queue
q = Queue.LifoQueue()for i in range(5):
q.put(i)while not q.empty(): print q.get()
输出：
4
3
2
1
可以看到仅仅是将Queue.Quenu类替换为Queue.LifiQueue类
优先级队列
class Queue.PriorityQueue(maxsize=0)
构造一个优先队列。maxsize用法同上。
import Queueimport threadingclass Job(object):
def __init__(self, priority, description):
self.priority = priority
self.description = description print 'Job:',description return
def __cmp__(self, other):
return cmp(self.priority, other.priority)
q = Queue.PriorityQueue()
q.put(Job(3, 'level 3 job'))
q.put(Job(10, 'level 10 job'))
q.put(Job(1, 'level 1 job'))def process_job(q):
while True:
next_job = q.get() print 'for:', next_job.description
q.task_done()
workers = [threading.Thread(target=process_job, args=(q,)),
threading.Thread(target=process_job, args=(q,))
]for w in workers:
w.setDaemon(True)
w.start()
q.join()
结果
Job: level 3 jobJob: level 10 jobJob: level 1 jobfor: level 1 jobfor: level 3 jobfor: job: level 10 job
一些常用方法
task_done()
意味着之前入队的一个任务已经完成。由队列的消费者线程调用。每一个get()调用得到一个任务，接下来的task_done()调用告诉队列该任务已经处理完毕。
如果当前一个join()正在阻塞，它将在队列中的所有任务都处理完时恢复执行（即每一个由put()调用入队的任务都有一个对应的task_done()调用）。
join()
阻塞调用线程，直到队列中的所有任务被处理掉。
只要有数据被加入队列，未完成的任务数就会增加。当消费者线程调用task_done()（意味着有消费者取得任务并完成任务），未完成的任务数就会减少。当未完成的任务数降到0，join()解除阻塞。
put(item[, block[, timeout]])
将item放入队列中。
如果可选的参数block为True且timeout为空对象（默认的情况，阻塞调用，无超时）。
如果timeout是个正整数，阻塞调用进程最多timeout秒，如果一直无空空间可用，抛出Full异常（带超时的阻塞调用）。
如果block为False，如果有空闲空间可用将数据放入队列，否则立即抛出Full异常
其非阻塞版本为put_nowait等同于put(item, False)
get([block[, timeout]])
从队列中移除并返回一个数据。block跟timeout参数同put方法
其非阻塞方法为｀get_nowait()｀相当与get(False)
empty()
如果队列为空，返回True,反之返回False
Queue 叫队列，是数据结构中的一种，基本上所有成熟的编程语言都内置了对 Queue 的支持。
Python 中的 Queue 模块实现了多生产者和多消费者模型，当需要在多线程编程中非常实用。而且该模块中的 Queue 类实现了锁原语，不需要再考虑多线程安全问题。
该模块内置了三种类型的 Queue，分别是 class queue.Queue(maxsize=0) ， class queue.LifoQueue(maxsize=0) 和 class queue.PriorityQueue(maxsize=0) 。它们三个的区别仅仅是取出时的顺序不一致而已。
Queue 是一个 FIFO 队列，任务按照添加的顺序被取出。
LifoQueue 是一个 LIFO 队列，类似堆栈，后添加的任务先被取出。
PriorityQueue 是一个优先级队列，队列里面的任务按照优先级排序，优先级高的先被取出。
如你所见，就是上面所说的三种不同类型的内置队列，其中 maxsize 是个整数，用于设置可以放入队列中的任务数的上限。当达到这个大小的时候，插入操作将阻塞至队列中的任务被消费掉。如果 maxsize 小于等于零，则队列尺寸为无限大。
向队列中添加任务，直接调用 put() 函数即可
put() 函数完整的函数签名如下 Queue.put(item, block=True, timeout=None) ，如你所见，该函数有两个可选参数。
默认情况下，在队列满时，该函数会一直阻塞，直到队列中有空余的位置可以添加任务为止。如果 timeout 是正数，则最多阻塞 timeout 秒，如果这段时间内还没有空余的位置出来，则会引发 Full 异常。
当 block 为 false 时，timeout 参数将失效。同时如果队列中没有空余的位置可添加任务则会引发 Full 异常，否则会直接把任务放入队列并返回，不会阻塞。
另外，还可以通过 Queue.put_nowait(item) 来添加任务，相当于 Queue.put(item, False) ，不再赘述。同样，在队列满时，该操作会引发 Full 异常。
从队列中获取任务，直接调用 get() 函数即可。
与 put() 函数一样， get() 函数也有两个可选参数，完整签名如下 Queue.get(block=True, timeout=None) 。
默认情况下，当队列空时调用该函数会一直阻塞，直到队列中有任务可获取为止。如果 timeout 是正数，则最多阻塞 timeout 秒，如果这段时间内还没有任务可获取，则会引发 Empty 异常。
当 block 为 false 时，timeout 参数将失效。同时如果队列中没有任务可获取则会立刻引发 Empty 异常，否则会直接获取一个任务并返回，不会阻塞。
另外，还可以通过 Queue.get_nowait() 来获取任务，相当于 Queue.get(False) ，不再赘述。同样，在队列为空时，该操作会引发 Empty 异常。
Queue.qsize() 函数返回队列的大小。注意这个大小不是精确的，qsize() 0 不保证后续的 get() 不被阻塞，同样 qsize() maxsize 也不保证 put() 不被阻塞。
如果队列为空，返回 True ，否则返回 False 。如果 empty() 返回 True ，不保证后续调用的 put() 不被阻塞。类似的，如果 empty() 返回 False ，也不保证后续调用的 get() 不被阻塞。
如果队列是满的返回 True ，否则返回 False 。如果 full() 返回 True 不保证后续调用的 get() 不被阻塞。类似的，如果 full() 返回 False 也不保证后续调用的 put() 不被阻塞。
queue.Queue() 是 FIFO 队列，出队顺序跟入队顺序是一致的。
queue.LifoQueue() 是 LIFO 队列，出队顺序跟入队顺序是完全相反的，类似于栈。
优先级队列中的任务顺序跟放入时的顺序是无关的，而是按照任务的大小来排序，最小值先被取出。那任务比较大小的规则是怎么样的呢。
注意，因为列表的比较对规则是按照下标顺序来比较的，所以在没有比较出大小之前 ，队列中所有列表对应下标位置的元素类型要一致。
好比 [2,1] 和 ["1","b"] 因为第一个位置的元素类型不一样，所以是没有办法比较大小的，所以也就放入不了优先级队列。
然而对于 [2,1] 和 [1,"b"] 来说即使第二个元素的类型不一致也是可以放入优先级队列的，因为只需要比较第一个位置元素的大小就可以比较出结果了，就不需要比较第二个位置元素的大小了。
但是对于 [2,1] 和 1 [2,"b"] 来说，则同样不可以放入优先级队列，因为需要比较第二个位置的元素才可以比较出结果，然而第二个位置的元素类型是不一致的，无法比较大小。
综上，也就是说， 直到在比较出结果之前，对应下标位置的元素类型都是需要一致的 。
下面我们自定义一个动物类型，希望按照年龄大小来做优先级排序。年龄越小优先级越高。
本章节介绍了队列以及其常用操作。因为队列默认实现了锁原语，因此在多线程编程中就不需要再考虑多线程安全问题了，对于程序员来说相当友好了。
进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道(不推荐使用)，这两种方式都是使用消息传递的
创建队列的类（底层就是以管道和锁定的方式实现） ：
参数介绍：
方法介绍：
主要方法：
其他方法(了解)：
Python实现简单多线程任务队列
最近我在用梯度下降算法绘制神经网络的数据时，遇到了一些算法性能的问题。梯度下降算法的代码如下（伪代码）：
defgradient_descent(): # the gradient descent code plotly.write(X, Y)
一般来说，当网络请求 plot.ly 绘图时会阻塞等待返回，于是也会影响到其他的梯度下降函数的执行速度。
一种解决办法是每调用一次 plotly.write 函数就开启一个新的线程，但是这种方法感觉不是很好。 我不想用一个像 cerely（一种分布式任务队列）一样大而全的任务队列框架，因为框架对于我的这点需求来说太重了，并且我的绘图也并不需要 redis 来持久化数据。
那用什么办法解决呢？我在 python 中写了一个很小的任务队列，它可以在一个单独的线程中调用 plotly.write函数。下面是程序代码。
fromthreadingimportThreadimportQueueimporttime classTaskQueue(Queue.Queue):
首先我们继承 Queue.Queue 类。从 Queue.Queue 类可以继承 get 和 put 方法，以及队列的行为。
def__init__(self, num_workers=1): Queue.Queue.__init__(self) self.num_workers=num_workers self.start_workers()
初始化的时候，我们可以不用考虑工作线程的数量。
defadd_task(self, task,*args,**kwargs): args=argsor() kwargs=kwargsor{} self.put((task, args, kwargs))
我们把 task, args, kwargs 以元组的形式存储在队列中。*args 可以传递数量不等的参数，**kwargs 可以传递命名参数。
defstart_workers(self): foriinrange(self.num_workers): t=Thread(target=self.worker) t.daemon=True t.start()
我们为每个 worker 创建一个线程，然后在后台删除。
下面是 worker 函数的代码：
defworker(self): whileTrue: tupl=self.get() item, args, kwargs=self.get() item(*args,**kwargs) self.task_done()
worker 函数获取队列顶端的任务，并根据输入参数运行，除此之外，没有其他的功能。下面是队列的代码：
我们可以通过下面的代码测试：
defblokkah(*args,**kwargs): time.sleep(5) print“Blokkah mofo!” q=TaskQueue(num_workers=5) foriteminrange(1): q.add_task(blokkah) q.join()# wait for all the tasks to finish. print“Alldone!”
Blokkah 是我们要做的任务名称。队列已经缓存在内存中，并且没有执行很多任务。下面的步骤是把主队列当做单独的进程来运行，这样主程序退出以及执行数据库持久化时，队列任务不会停止运行。但是这个例子很好地展示了如何从一个很简单的小任务写成像工作队列这样复杂的程序。
defgradient_descent(): # the gradient descent code queue.add_task(plotly.write, x=X, y=Y)
修改之后，我的梯度下降算法工作效率似乎更高了。如果你很感兴趣的话，可以参考下面的代码。fromthreadingimportThreadimportQueueimporttime classTaskQueue(Queue.Queue): def__init__(self, num_workers=1):Queue.Queue.__init__(self)self.num_workers=num_workersself.start_workers() defadd_task(self, task,*args,**kwargs):args=argsor()kwargs=kwargsor{}self.put((task, args, kwargs)) defstart_workers(self):foriinrange(self.num_workers):t=Thread(target=self.worker)t.daemon=Truet.start() defworker(self):whileTrue:tupl=self.get()item, args, kwargs=self.get()item(*args,**kwargs)self.task_done() deftests():defblokkah(*args,**kwargs):time.sleep(5)print"Blokkah mofo!" q=TaskQueue(num_workers=5) foriteminrange(10):q.add_task(blokkah) q.join()# block until all tasks are doneprint"All done!" if__name__=="__main__":tests()
重庆分公司
重庆分公司
