Python使用线程队列+实现线程池示例 - 行业资讯 - 肥雀云

<强> 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自己构建一个线程池

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

Queue.qsize():返回队列的大小。

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

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

队列。把(项目,块=True,超时=None):往队列里放数据。

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

队列。得到(项目,块=True,超时=None):从队列里取数据。

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

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

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

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

　　　　　　进口队列　　进口线程　　导入的时间　　　　队列=Queue.Queue () 　　　　类ThreadNum (threading.Thread): 　　def __init__(自我、队列): 　　threading.Thread.__init__(自我) 　　自我。队列=队列　　　　def运行(自我): 　　而真正的: 　　#消费者端,从队列中获取num 　　num=self.queue.get () 　　打印(“检索”,num) 　　time . sleep (1) 　　#在完成这项工作之后,使用queue.task_done()函数向任务已经完成的队列发送一个信号　　self.queue.task_done () 　　　　打印(完成“消费者”) 　　　　def main (): 　　#产生一个线程池,并把消息传递给线程函数进行处理,这里开启10个并发　　因为我在范围(5): 　　t=ThreadNum(队列) 　　t.setDaemon(真正的) 　　t.start () 　　　　#往队列中填数据　　num的范围(10): 　　queue.put (num) 　　#等> 　　(“检索”,0) 　　(1)“检索”(“检索”,2) 　　(“检索”,3) 　　(4)“检索” 　　(5)“检索”(“检索”,6) 　　(7)“检索” 　　(“检索”,8) 　　(“检索”,9) 　　　　

<强>具体工作步骤描述如下:

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

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

3,生成守护线程池。

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

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

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

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

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

　　　　　　进口队列　　进口线程　　导入的时间　　　　队列=Queue.Queue () 　　　　类ThreadNum (threading.Thread): 　　”“没打印一个数字等待1秒,并发打印10个数字需要多少秒吗?”“ 　　def __init__(自我、队列): 　　threading.Thread.__init__(自我) 　　自我。队列=队列　　　　def运行(自我): 　　做=False 　　而没有完成: 　　#消费者端,从队列中获取num 　　num=self.queue.get () 　　如果num没有: 　　做=True 　　其他: 　　打印(“检索”,num) 　　time . sleep (1) 　　#在完成这项工作之后,使用queue.task_done()函数向任务已经完成的队列发送一个信号　　self.queue.task_done () 　　　　打印(完成“消费者”) 　　def main (): 　　#产生一个线程池,并把消息传递给线程函数进行处理,这里开启10个并发　　因为我在范围(5): 　　t=ThreadNum(队列) 　　t.setDaemon(真正的) 　　t.start () 　　　　#往队列中填错数据　　num的范围(10): 　　queue.put (num) 　　　　queue.join () 　　time . sleep (100) 　　因为我在范围(10): 　　queue.put(没有) 　　打印(“无”) 　　time . sleep (200) 　　　　if __name__==癬_main__”: 　　开始=time.time () 　　main () 　　打印”运行时间:% s“% (time.time()——开始)