伏龙阁资源网

Python 使用threading+Queue实现线程池示例
站长资源 2024/11/25 佚名
2 0

一、线程池

1、为什么需要使用线程池

1.1 创建/销毁线程伴随着系统开销，过于频繁的创建/销毁线程，会很大程度上影响处理效率。

记创建线程消耗时间T1，执行任务消耗时间T2，销毁线程消耗时间T3，如果T1+T3>T2，那说明开启一个线程来执行这个任务太不划算了！在线程池缓存线程可用已有的闲置线程来执行新任务，避免了创建/销毁带来的系统开销。

1.2 线程并发数量过多，抢占系统资源从而导致阻塞。

线程能共享系统资源，如果同时执行的线程过多，就有可能导致系统资源不足而产生阻塞的情况。

1.3 对线程进行一些简单的管理。

比如：延时执行、定时循环执行的策略等，运用线程池都能进行很好的实现。

2、Python中建立线程池的方法

2.1 使用threadpool模块，这是个python的第三方模块，支持python2和python3

2.2 使用concurrent.futures模块，这个模块是python3中自带的模块，python2.7以上版本也可以安装使用

2.3 自己构建一个线程池

二、队列（queue）

Queue模块提供的队列（FIFO）适用于多线程编程，在生产者(producer)和消费者(consumer)之间线程安全(thread-safe)地传递消息或其它数据，因此多个线程可以共用同一个Queue实例。常用方法:

Queue.qsize()：返回queue的大小。

Queue.empty():判断队列是否为空，通常不太靠谱。

Queue.full():判断是否满了。

Queue.put(item, block=True, timeout=None): 往队列里放数据。

Queue.put_nowait(item):往队列里存放元素，不等待

Queue.get(item, block=True, timeout=None): 从队列里取数据。

Queue.get_nowait(item):从队列里取元素，不等待

Queue.task_done()：表示队列中某个元素是否的使用情况，使用结束会发送信息。

Queue.join()：一直阻塞直到队列中的所有元素都执行完毕。

三、使用threading+Queue处理多任务

假设有十个任务需要处理，打算在后台开启五个线程，简化后的模型

import Queue
import threading
import time
 
queue = Queue.Queue()
 
class ThreadNum(threading.Thread):
  def __init__(self, queue):
    threading.Thread.__init__(self)
    self.queue = queue
 
  def run(self):
    while True:
      #消费者端，从队列中获取num
      num = self.queue.get()
      print("Retrieved", num)
      time.sleep(1) 
      #在完成这项工作之后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号
      self.queue.task_done()
    
    print("Consumer Finished")
 
def main():
  #产生一个 threads pool, 并把消息传递给thread函数进行处理，这里开启10个并发
  for i in range(5):
    t = ThreadNum(queue)
    t.setDaemon(True)
    t.start()
  
  #往队列中填数据 
  for num in range(10):
    queue.put(num)
    #wait on the queue until everything has been processed
  
  queue.join()
   
if __name__ == '__main__':
  main()
  time.sleep(500)

输出为：

('Retrieved', 0)
 ('Retrieved', 1)('Retrieved', 2)
('Retrieved', 3)
('Retrieved', 4)
('Retrieved', 5)('Retrieved', 6)
('Retrieved', 7)
('Retrieved', 8)
 ('Retrieved', 9)

具体工作步骤描述如下：

1、创建一个 Queue.Queue() 的实例，然后使用数据对它进行填充。

2、将经过填充数据的实例传递给线程类，后者是通过继承 threading.Thread 的方式创建的。

3、生成守护线程池。

4、每次从队列中取出一个项目，并使用该线程中的数据和 run 方法以执行相应的工作。

5、在完成这项工作之后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号。

6、对队列执行 join 操作，实际上意味着等到队列为空，再退出主程序。

在使用这个模式时需要注意一点：通过将守护线程设置为 true，程序运行完自动退出。好处是在退出之前，可以对队列执行 join 操作、或者等到队列为空。

注意运行main函数后继续执行time.sleep(500)，可以观察到主线程未结束的情况下ThreadNum(queue)生成的线程还在运行。如果需要停止线程的话可以对以上代码加以修改。

import Queue
import threading
import time
 
queue = Queue.Queue()
 
class ThreadNum(threading.Thread):
  """没打印一个数字等待1秒，并发打印10个数字需要多少秒？"""
  def __init__(self, queue):
    threading.Thread.__init__(self)
    self.queue = queue
 
  def run(self):
    done = False
    while not done:
      #消费者端，从队列中获取num
      num = self.queue.get()
      if num is None:
        done = True
      else:
        print("Retrieved", num)
      time.sleep(1) 
      #在完成这项工作之后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号
      self.queue.task_done()
    
    print("Consumer Finished")
def main():
  #产生一个 threads pool, 并把消息传递给thread函数进行处理，这里开启10个并发
  for i in range(5):
    t = ThreadNum(queue)
    t.setDaemon(True)
    t.start()
  
  #往队列中填错数据 
  for num in range(10):
    queue.put(num)
  
  queue.join()
  time.sleep(100)
  for i in range(10):
    queue.put(None)
    print('None')
  time.sleep(200)
   
if __name__ == '__main__':
  start = time.time()
  main()
  print"Elapsed Time: %s" % (time.time() - start)

main函数执行完后队列向线程发送None消息，触发线程的停止标识，这样就可以动态管理线程池了。

以上这篇Python 使用threading+Queue实现线程池示例就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持。

广告合作：本站广告合作请联系QQ：858582 申请时备注：广告合作（否则不回）
免责声明：本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除！
www.wwsws.com 伏龙阁资源网
39,976影音资源
44,792技术资源
21,817软件资源
651,128站长资源

更新日志

友情链接

杰晶网络 DDR爱好者之家 桃源资源网 杰网资源 富贵资源网 南强小屋 铁雪资源网 幽灵资源网 万梅资源网 狼山资源网 白云岛资源网 昆仑资源网 相思资源网 明霞山资源网 内蒙古资源网 黑松山资源网 茶园资源网 饿虎岗资源网 大旗谷资源网 常春岛资源网 岱庙资源网 兴国资源网 快活林资源网 蝙蝠岛资源网 帝王谷资源网 白云城资源网 伏龙阁资源网 清风细雨楼 天枫庄资源网 圆月山庄资源网 无争山庄资源网 神水资源网 移花宫资源网 神剑山庄资源网 无为清净楼资源网 金钱帮资源网 丐帮资源网 华山资源网 极乐门资源网 小李飞刀资源网 凤求凰客栈 风云阁资源网 金狮镖局 鸳鸯亭资源网 千金楼资源网 更多链接