Python ThreadPoolExecutor线程池
概念
Python
中已经有了threading
模块,为什么还需要线程池呢,线程池又是什么东西呢?以爬虫为例,需要控制同时爬取的线程数,例子中创建了20个线程,而同时只允许3个线程在运行,但是20个线程都需要创建和销毁,线程的创建是需要消耗系统资源的,有没有更好的方案呢?其实只需要三个线程就行了,每个线程各分配一个任务,剩下的任务排队等待,当某个线程完成了任务的时候,排队任务就可以安排给这个线程继续执行。
这就是线程池的思想(当然没这么简单),但是自己编写线程池很难写的比较完美,还需要考虑复杂情况下的线程同步,很容易发生死锁。从Python3.2
开始,标准库为我们提供了concurrent.futures
模块,它提供了ThreadPoolExecutor
和ProcessPoolExecutor
两个类,实现了对threading
和multiprocessing
的进一步抽象(这里主要关注线程池),不仅可以帮我们自动调度线程,还可以做到:
- 主线程可以获取某一个线程(或者任务的)的状态,以及返回值。
- 当一个线程完成的时候,主线程能够立即知道。
- 让多线程和多进程的编码接口一致。
实例
简单使用
from concurrent.futures import ThreadPoolExecutor
import time# 参数times用来模拟网络请求的时间
def get_html(times):time.sleep(times)print("get page {}s finished".format(times))return timesexecutor = ThreadPoolExecutor(max_workers=2)
# 通过submit函数提交执行的函数到线程池中,submit函数立即返回,不阻塞
task1 = executor.submit(get_html, (3))
task2 = executor.submit(get_html, (2))
# done方法用于判定某个任务是否完成
print(task1.done())
# cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功
print(task2.cancel())
time.sleep(4)
print(task1.done())
# result方法可以获取task的执行结果
print(task1.result())# 执行结果
# False # 表明task1未执行完成
# False # 表明task2取消失败,因为已经放入了线程池中
# get page 2s finished
# get page 3s finished
# True # 由于在get page 3s finished之后才打印,所以此时task1必然完成了
# 3 # 得到task1的任务返回值
ThreadPoolExecutor
构造实例的时候,传入max_workers
参数来设置线程池中最多能同时运行的线程数目。- 使用
submit
函数来提交线程需要执行的任务(函数名和参数)到线程池中,并返回该任务的句柄(类似于文件、画图),注意submit()
不是阻塞的,而是立即返回。- 通过
submit
函数返回的任务句柄,能够使用done()
方法判断该任务是否结束。上面的例子可以看出,由于任务有2s的延时,在task1
提交后立刻判断,task1
还未完成,而在延时4s之后判断,task1
就完成了。- 使用
cancel()
方法可以取消提交的任务,如果任务已经在线程池中运行了,就取消不了。这个例子中,线程池的大小设置为2,任务已经在运行了,所以取消失败。如果改变线程池的大小为1,那么先提交的是task1
,task2
还在排队等候,这是时候就可以成功取消。- 使用
result()
方法可以获取任务的返回值。查看内部代码,发现这个方法是阻塞的。
as_completed
上面虽然提供了判断任务是否结束的方法,但是不能在主线程中一直判断啊。有时候我们是得知某个任务结束了,就去获取结果,而不是一直判断每个任务有没有结束。这是就可以使用as_completed
方法一次取出所有任务的结果。
from concurrent.futures import ThreadPoolExecutor, as_completed
import time# 参数times用来模拟网络请求的时间
def get_html(times):time.sleep(times)print("get page {}s finished".format(times))return timesexecutor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]for future in as_completed(all_task):data = future.result()print("in main: get page {}s success".format(data))# 执行结果
# get page 2s finished
# in main: get page 2s success
# get page 3s finished
# in main: get page 3s success
# get page 4s finished
# in main: get page 4s success
as_completed()
方法是一个生成器,在没有任务完成的时候,会阻塞,在有某个任务完成的时候,会yield
这个任务,就能执行for循环下面的语句,然后继续阻塞住,循环到所有的任务结束。从结果也可以看出,先完成的任务会先通知主线程。
map
除了上面的as_completed
方法,还可以使用executor.map
方法,但是有一点不同。
from concurrent.futures import ThreadPoolExecutor
import time# 参数times用来模拟网络请求的时间
def get_html(times):time.sleep(times)print("get page {}s finished".format(times))return timesexecutor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的urlfor data in executor.map(get_html, urls):print("in main: get page {}s success".format(data))
# 执行结果
# get page 2s finished
# get page 3s finished
# in main: get page 3s success
# in main: get page 2s success
# get page 4s finished
# in main: get page 4s success
使用map
方法,无需提前使用submit
方法,map
方法与python
标准库中的map
含义相同,都是将序列中的每个元素都执行同一个函数。上面的代码就是对urls
的每个元素都执行get_html
函数,并分配各线程池。可以看到执行结果与上面的as_completed
方法的结果不同,输出顺序和urls
列表的顺序相同,就算2s的任务先执行完成,也会先打印出3s的任务先完成,再打印2s的任务完成。
wait
wait
方法可以让主线程阻塞,直到满足设定的要求。
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time# 参数times用来模拟网络请求的时间
def get_html(times):time.sleep(times)print("get page {}s finished".format(times))return timesexecutor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=ALL_COMPLETED)
print("main")
# 执行结果
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main
wait
方法接收3个参数,等待的任务序列、超时时间以及等待条件。等待条件return_when
默认为ALL_COMPLETED
,表明要等待所有的任务都结束。可以看到运行结果中,确实是所有任务都完成了,主线程才打印出main
。等待条件还可以设置为FIRST_COMPLETED
,表示第一个任务完成就停止等待。
源码分析
cocurrent.future
模块中的future
的意思是未来对象,可以把它理解为一个在未来完成的操作,这是异步编程的基础 。在线程池submit()
之后,返回的就是这个future
对象,返回的时候任务并没有完成,但会在将来完成。也可以称之为task的返回容器,这个里面会存储task的结果和状态。那ThreadPoolExecutor
内部是如何操作这个对象的呢?
下面简单介绍ThreadPoolExecutor
的部分代码:
1.init方法
init
方法中主要重要的就是任务队列和线程集合,在其他方法中需要使用到。
ThreadPoolExecutor_init方法.png
2.submit方法
submit
中有两个重要的对象,_base.Future()
和_WorkItem()
对象,_WorkItem()
对象负责运行任务和对future
对象进行设置,最后会将future
对象返回,可以看到整个过程是立即返回的,没有阻塞。
ThreadPoolExecutor_submit方法.png
3.adjust_thread_count方法
这个方法的含义很好理解,主要是创建指定的线程数。但是实现上有点难以理解,比如线程执行函数中的weakref.ref,涉及到了弱引用等概念,留待以后理解。
ThreadPoolExecutor_adjust_thread_count方法.png
4._WorkItem对象
_WorkItem
对象的职责就是执行任务和设置结果。这里面主要复杂的还是self.future.set_result(result)
。
_WorkItem类.png
5.线程执行函数--_worker
这是线程池创建线程时指定的函数入口,主要是从队列中依次取出task执行,但是函数的第一个参数还不是很明白。留待以后。
ThreadPoolExecutor_线程执行函数_worker.png
总结
- future的设计理念很棒,在线程池/进程池和携程中都存在future对象,是异步编程的核心。
- ThreadPoolExecutor 让线程的使用更加方便,减小了线程创建/销毁的资源损耗,无需考虑线程间的复杂同步,方便主线程与子线程的交互。
- 线程池的抽象程度很高,多线程和多进程的编码接口一致。
Python ThreadPoolExecutor线程池相关推荐
- python ThreadPoolExecutor线程池(实例)
初识 Python中已经有了threading模块,为什么还需要线程池呢,线程池又是什么东西呢?在介绍线程同步的信号量机制的时候,举得例子是爬虫的例子,需要控制同时爬取的线程数,例子中创建了20个线程 ...
- Python运用线程池ThreadPoolExecutor多线程下载搜索图片
用chrome解析某搜索网站图片搜索请求包,再参考网上一些已有的代码,在vscode+python3.7环境,多线程下载搜索结果页面上图片的代码,其中加入了键盘中止和ThreadPoolExecuto ...
- Python实现线程池
Python实现线程池 最近在做一些文本处理方面的事情,考虑到程序利用并发性可以提高执行效率(不纠结特殊反例),于是入围的Idea如使用多进程或多线程达到期望的目标,对于进程或线程的创建是有代价的,那 ...
- ThreadPoolExecutor线程池,shutdown和shutdownNow关闭线程池方式对比,以及确保线程池能够彻底关闭的一种方式
1. ThreadPoolExecutor线程池 1.1 创建线程池,构造方法的几个参数说明及创建如下. 1.2 shutdown方式关闭线程池 a. 空闲且能interrupt表示该线程处于阻塞等待 ...
- 13.ThreadPoolExecutor线程池之submit方法
jdk1.7.0_79 在上一篇<ThreadPoolExecutor线程池原理及其execute方法>中提到了线程池ThreadPoolExecutor的原理以及它的execute方法 ...
- Java Executor源码解析(3)—ThreadPoolExecutor线程池execute核心方法源码【一万字】
基于JDK1.8详细介绍了ThreadPoolExecutor线程池的execute方法源码! 上一篇文章中,我们介绍了:Java Executor源码解析(2)-ThreadPoolExecutor ...
- ThreadPoolExecutor线程池核心参数详解
理解ThreadPoolExecutor线程池的corePoolSize.maximumPoolSize和poolSize 我们知道,受限于硬件.内存和性能,我们不可能无限制的创建任意数量的线程,因为 ...
- 手写一个线程池,带你学习ThreadPoolExecutor线程池实现原理
摘要:从手写线程池开始,逐步的分析这些代码在Java的线程池中是如何实现的. 本文分享自华为云社区<手写线程池,对照学习ThreadPoolExecutor线程池实现原理!>,作者:小傅哥 ...
- ThreadPoolExecutor线程池原理
ThreadPoolExecutor线程池原理 线程池原理 1. 线程池的简单介绍 1.1 线程池是什么 1.2 线程池解决的核心问题是什么 2. 线程池的实现原理 2.1 线程池的执行流程 2.2 ...
最新文章
- 用C#编写获取远程IP,MAC的方法
- 硬核!30 张图解 HTTP 常见的面试题
- Jenkins的环境部署
- 存储时间:高清3D化,技术应用化?
- mysql not in语句_MySQL命令
- JAVA生产环境验证_Java生产环境下性能监控与调优详解
- php将字符变为数字,数字字符怎么转化为数字 php 怎么将字符转成数字
- html上传文件是否合法,java 在前端页面上传图片文件,验证是否为合法的图片
- 纯CSS实现table表头固定(自创备忘)
- XDeepFM 模型,字节跳动短视频内容理解和推荐系统
- DBGrid 应用全书
- 计算机应用综合实践课程报告,南通大学继续教育学院软件工程综合实践课程报告书——生产信息化工作管理系统...
- oracle查询所有表名_oracle删错数据了,要跑路吗,等一下,先抢救一下
- PWM DAC vs. Standalone
- ARM寄存器的操作方法
- 北华大学c语言题库百度云,北华大学C语言题库精简打印版(全).docx
- Amaze UI 图标查询
- 芝麻授权 java调用_手机浏览器怎么调用支付宝进行用户授权呢?
- mysql如何获取当前时间_mysql怎么获取当前时间
- 银行会计学3(特点、核算的前提假设、对象和标准)
热门文章
- [android]不解锁刷机
- Android 音频源码分析——AudioTrack设备选择
- 1:ul居中文 2:字居中 3:去除ul和li前面的点
- Android平台研发人才缺口30万
- ChatGPT大模型用于刑侦破案只能是虚构故事吗?AI大模型企业服务趋势探讨(一)...
- 成功解决E: 无法获得锁 /var/lib/dpkg/lock-frontend - open (11: 资源暂时不可用) E: 无法获取 dpkg 前端锁(/var/lib/dpkg/lock
- Linux中使用RAID技术提升磁盘读写速度及数据安全
- 移动端开发应注意的事项有哪些?
- PCB中大面积敷铜的目的
- 实时级嵌入式系统半实物仿真测试平台技术指标