Python 使用threading+Queue实现线程池示例
一、线程池
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实现线程池示例就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持【听图阁-专注于Python设计】。