Python线程编程—同步队列

Python线程编程—同步队列

咱们常常会采用生产者/消费者关系的两个线程来处理一个共享缓冲区的数据。例如一 个生产者线程接受用户数据放入一个共享缓冲区里，等待一个消费者线程对数据取出处理。可是若是缓冲区的过小而生产者和消费者两个异步线程的速度不一样时，容 易出现一个线程等待另外一个状况。为了尽量的缩短共享资源并以相同速度工做的各线程的等待时间，咱们能够使用一个“队列”来提供额外的缓冲区。

建立一个“队列”对象

import Queue
myqueue = Queue.Queue(maxsize = 10)

Queue.Queue类便是一个队列的同步实现。队列长度可为无限或者有限。可经过Queue的构造函数的可选参数maxsize来设定队列长度。若是maxsize小于1就表示队列长度无限。

将一个值放入队列中

myqueue.put(10)

调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数， 默认为1。若是队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。若是block为0，put方法将引起Full异 常。

将一个值从队列中取出

myqueue.get()

调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为1。若是队列为空且block为1，get()就使调用线程暂停，直至有项目可用。若是block为0，队列将引起Empty异常。

咱们用一个例子来展现如何使用Queue

#!/usr/bin/env python import Queueimport threadingimport urllib2import time
          
hosts = ["http://yahoo.com", "http://google.com.hk", "http://amazon.com",
          "http://ibm.com", "http://apple.com"]
          
queue = Queue.Queue()
          class ThreadUrl(threading.Thread):
      """Threaded Url Grab"""        def __init__(self, queue):
              threading.Thread.__init__(self)
              self.queue = queue
          
       def run(self):
           while True:
                #grabs host from queue                 host = self.queue.get()     
                #grabs urls of hosts and prints first 1024 bytes of page                 url = urllib2.urlopen(host)
                print url.read(1024)        
                #signals to queue job is done                 self.queue.task_done()
   
 start = time.time()def main():
     #spawn a pool of threads, and pass them queue instance       for i in range(5):
            t = ThreadUrl(queue)
            t.setDaemon(True)
            t.start()
              
      #populate queue with data        for host in hosts:
            queue.put(host)
       #wait on the queue until everything has been processed         queue.join()
          
main()print "Elapsed Time: %s" % (time.time() - start)

 

在 Python 中使用线程时，这个模式是一种很常见的而且推荐使用的方式。具体工做步骤描述以下：

1. 建立一个 Queue.Queue() 的实例，而后使用数据对它进行填充。
2. 将通过填充数据的实例传递给线程类，后者是经过继承 threading.Thread 的方式建立的。
3. 生成守护线程池。
4. 每次从队列中取出一个项目，并使用该线程中的数据和 run 方法以执行相应的工做。
5. 在完成这项工做以后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号。
6. 对队列执行 join 操做，实际上意味着等到队列为空，再退出主程序。

在使用这个模式时须要注意一点：经过将守护线程设置为 true，将容许主线程或者程序仅在守护线程处于活动状态时才可以退出。这种方式建立了一种简单的方式以控制程序流程，由于在退出以前，您能够对队列执行 join 操做、或者等到队列为空。

join()

保持阻塞状态，直处处理了队列中的全部项目为止。在将一个项目添加到该队列时，未完成的任务的总数就会增长。当使用者线程调用 task_done() 以表示检索了该项目、并完成了全部的工做时，那么未完成的任务的总数就会减小。当未完成的任务的总数减小到零时，join() 就会结束阻塞状态。