python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池
作者：??HZ在掘金???? 发布时间：2022-08-28 08:26:45
标签：python,ThreadPoolExecutor,线程池,ProcessPoolExecutor,进程池
1、ThreadPoolExecutor多线程
<1>为什么需要线程池呢?
对于io密集型，提高执行的效率。
线程的创建是需要消耗系统资源的。
所以线程池的思想就是：每个线程各自分配一个任务，剩下的任务排队等待，当某个线程完成了任务的时候，排队任务就可以安排给这个线程继续执行。
<2>标准库concurrent.futures模块
它提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，
分别实现了对threading模块和multiprocessing模块的进一步抽象。
不仅可以帮我们自动调度线程，还可以做到:
主线程可以获取某一个线程(或者任务)的状态，以及返回值
当一个线程完成的时候，主线程能够立即知道
让多线程和多进程的编码接口一致
<3>简单使用
# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor
import time
# 参数times用来模拟网络请求时间
def get_html(times):
print("get page {}s finished".format(times))
return times
# 创建线程池
# 设置线程池中最多能同时运行的线程数目，其他等待
executor = ThreadPoolExecutor(max_workers=2)
# 通过submit函数提交执行的函数到线程池中，submit函数立即返回,不阻塞
# task1和task2是任务句柄
task1 = executor.submit( get_html, (2) )
task2 = executor.submit( get_html, (3) )
# done()方法用于判断某个任务是否完成，bool型，完成返回True，没有完成返回False
print( task1.done() )
# cancel()方法用于取消某个任务，该任务没有放到线程池中才能被取消，如果已经放进线程池子中，则不能被取消
# bool型，成功取消了返回True，没有取消返回False
print( task2.cancel() )
# result()方法可以获取task的执行结果，前提是get_html()函数有返回值
print( task1.result() )
print( task2.result() )
# 结果：
# get page 3s finished
# get page 2s finished
# True
# False
# 2
# 3
ThreadPoolExecutor类在构造实例的时候，传入max_workers参数来设置线程池中最多能同时运行的线程数目
使用submit()函数来提交线程需要执行任务(函数名和参数)到线程池中，并返回该任务的句柄，
注意：submit()不是阻塞的，而是立即返回。
通过submit()函数返回的任务句柄，能够使用done()方法判断该任务是否结束，使用cancel()方法来取消，使用result()方法可以获取任务的返回值，查看内部代码，发现该方法是阻塞的
<4>as_completed（一次性获取所有的结果）
上面虽然提供了判断任务是否结束的方法，但是不能在主线程中一直判断，有时候我们是得知某个任务结束了，就去获取结果，而不是一直判断每个任务有没有结束。这时候就可以使用as_completed方法一次取出所有任务的结果。
# -*-coding:utf-8 -*-
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 times
# 创建线程池子
# 设置最多2个线程运行，其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
# 一次性把所有的任务都放进线程池，得到一个句柄，但是最多只能同时执行2个任务
all_task = [ executor.submit(get_html,(each_url)) for each_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
# 从结果可以看到，并不是先传入哪个url，就先执行哪个url，没有先后顺序
<5>map()方法
除了上面的as_completed()方法，还可以使用execumap方法。但是有一点不同，使用map方法，不需提前使用submit方法，
map方法与python标准库中的map含义相同，都是将序列中的每个元素都执行同一个函数。上面的代码就是对urls列表中的每个元素都执行get_html()函数，并分配各线程池。可以看到执行结果与上面的as_completed方法的结果不同，输出顺序和urls列表的顺序相同，就算2s的任务先执行完成，也会先打印出3s的任务先完成，再打印2s的任务完成
# -*-coding:utf-8 -*-
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 times
# 创建线程池子
# 设置最多2个线程运行，其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
for result in executor.map(get_html, urls):
print("in main:get page {}s success".format(result))
结果：
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
<6>wait()方法
wait方法可以让主线程阻塞，直到满足设定的要求。wait方法接收3个参数，等待的任务序列、超时时间以及等待条件。
等待条件return_when默认为ALL_COMPLETED,表明要等待所有的任务都借宿。可以看到运行结果中，确实是所有任务都完成了，主线程才打印出main，等待条件还可以设置为FIRST_COMPLETED,表示第一个任务完成就停止等待。
超时时间参数可以不设置：
wait()方法和as_completed(), map()没有关系。不管你是用as_completed()，还是用map()方法，你都可以在执行主线程之前使用wait()。
as_completed()和map()是二选一的。
# -*-coding:utf-8 -*-
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 times
# 创建线程池子
# 设置最多2个线程运行，其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
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
2、ProcessPoolExecutor多进程
<1>同步调用方式: 调用，然后等返回值，能解耦，但是速度慢
import datetime
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import time, random, os
import requests
def task(name):
print('％s ％s is running'％(name,os.getpid()))
#print(datetime.datetime.now().strftime("％Y-％m-％d ％H:％M:％S"))
if __name__ == '__main__':
p = ProcessPoolExecutor(4) # 设置
for i in range(10):
# 同步调用方式，不仅要调用，还要等返回值
obj = p.submit(task, "进程pid：") # 传参方式(任务名，参数),参数使用位置或者关键字参数
res = obj.result()
p.shutdown(wait=True) # 关闭进程池的入口，等待池内任务运行结束
print("主")
################
################
# 另一个同步调用的demo
def get(url):
print('％s GET ％s' ％ (os.getpid(),url))
time.sleep(3)
response = requests.get(url)
if response.status_code == 200:
res = response.text
else:
res = "下载失败"
return res # 有返回值
def parse(res):
time.sleep(1)
print("％s 解析结果为％s" ％(os.getpid(),len(res)))
if __name__ == "__main__":
urls = [
'https://www.baidu.com',
'https://www.sina.com.cn',
'https://www.tmall.com',
'https://www.jd.com',
'https://www.python.org',
'https://www.openstack.org',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
]
p=ProcessPoolExecutor(9)
l=[]
start = time.time()
for url in urls:
future = p.submit(get,url) # 需要等结果，所以是同步调用
l.append(future)
# 关闭进程池，等所有的进程执行完毕
p.shutdown(wait=True)
for future in l:
parse(future.result())
print('完成时间:',time.time()-start)
#完成时间: 13.209137678146362
<2>异步调用方式：只调用，不等返回值，可能存在耦合，但是速度快
def task(name):
print("％s ％s is running" ％(name,os.getpid()))
time.sleep(random.randint(1,3))
if __name__ == '__main__':
p = ProcessPoolExecutor(4) # 设置进程池内进程
for i in range(10):
# 异步调用方式，只调用，不等返回值
p.submit(task,'进程pid：') # 传参方式(任务名,参数),参数使用位置参数或者关键字参数
p.shutdown(wait=True) # 关闭进程池的入口，等待池内任务运行结束
print('主')
##################
##################
# 另一个异步调用的demo
def get(url):
print('％s GET ％s' ％ (os.getpid(),url))
time.sleep(3)
reponse = requests.get(url)
if reponse.status_code == 200:
res = reponse.text
else:
res = "下载失败"
parse(res) # 没有返回值
def parse(res):
time.sleep(1)
print('％s 解析结果为％s' ％(os.getpid(),len(res)))
if __name__ == '__main__':
urls = [
'https://www.baidu.com',
'https://www.sina.com.cn',
'https://www.tmall.com',
'https://www.jd.com',
'https://www.python.org',
'https://www.openstack.org',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
]
p = ProcessPoolExecutor(9)
start = time.time()
for url in urls:
future = p.submit(get,url)
p.shutdown(wait=True)
print("完成时间",time.time()-start)# 完成时间 6.293345212936401
<3>怎么使用异步调用方式，但同时避免耦合的问题？
(1)进程池：异步 + 回调函数，，cpu密集型，同时执行，每个进程有不同的解释器和内存空间，互不干扰
def get(url):
print('％s GET ％s' ％ (os.getpid(), url))
time.sleep(3)
response = requests.get(url)
if response.status_code == 200:
res = response.text
else:
res = '下载失败'
return res
def parse(future):
time.sleep(1)
# 传入的是个对象，获取返回值 需要进行result操作
res = future.result()
print("res",)
print('％s 解析结果为％s' ％ (os.getpid(), len(res)))
if __name__ == '__main__':
urls = [
'https://www.baidu.com',
'https://www.sina.com.cn',
'https://www.tmall.com',
'https://www.jd.com',
'https://www.python.org',
'https://www.openstack.org',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
]
p = ProcessPoolExecutor(9)
start = time.time()
for url in urls:
future = p.submit(get,url)
#模块内的回调函数方法，parse会使用future对象的返回值，对象返回值是执行任务的返回值
#回调应该是相当于parse(future)
future.add_done_callback(parse)
p.shutdown(wait=True)
print("完成时间",time.time()-start)#完成时间 33.79998469352722
(2)线程池：异步 + 回调函数，IO密集型主要使用方式,线程池:执行操作为谁有空谁执行
def get(url):
print("％s GET ％s" ％(current_thread().name,url))
time.sleep(3)
reponse = requests.get(url)
if reponse.status_code == 200:
res = reponse.text
else:
res = "下载失败"
return res
def parse(future):
time.sleep(1)
res = future.result()
print("％s 解析结果为％s" ％(current_thread().name,len(res)))
if __name__ == '__main__':
urls = [
'https://www.baidu.com',
'https://www.sina.com.cn',
'https://www.tmall.com',
'https://www.jd.com',
'https://www.python.org',
'https://www.openstack.org',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
]
p = ThreadPoolExecutor(4)
start = time.time()
for url in urls:
future = p.submit(get,url)
future.add_done_callback(parse)
p.shutdown(wait=True)
print("主",current_thread().name)
print("完成时间",time.time()-start)#完成时间 32.52604126930237
3、总结
1、线程不是越多越好，会涉及cpu上下文的切换（会把上一次的记录保存）。
2、进程比线程消耗资源，进程相当于一个工厂，工厂里有很多人，里面的人共同享受着福利资源，，一个进程里默认只有一个主线程，比如：开启程序是进程，里面执行的是线程，线程只是一个进程创建多个人同时去工作。
3、线程里有GIL全局解锁器：不允许cpu调度
4、计算密度型适用于多进程
5、线程：线程是计算机中工作的最小单元
6、进程：默认有主线程 (帮工作)可以多线程共存
7、协程：一个线程，一个进程做多个任务,使用进程中一个线程去做多个任务，微线程
8、GIL全局解释器锁：保证同一时刻只有一个线程被cpu调度
来源：https://juejin.cn/post/7078617200772251684
0
投稿猜你喜欢
1. 环境准备1.1 安装pillow 和 pytesseractpython模块库需要 pillow 和 pytesseract 这两个库- 前几天,在所有数据库服务器部署了监控磁盘空间的存储过程和作业后（MS SQL 监控磁盘空间告警），今天突然收到了两封告警邮件，好吧，存储规划
- 一、游戏玩法介绍：24点游戏是儿时玩的主要益智类游戏之一，玩法为：从一副扑克中抽取4张牌，对4张牌使用加减乘除中的任何方法，使计算结果为24
- 1、旅行商问题(Travelling salesman problem, TSP)旅行商问题是经典的组合优化问题，要求找到遍历所有城市且每个
- 安装pip install requests发送网络请求import requestsr=requests.get('http://
- 通过购物车的一个案列，把vuex学习了一篇。vuex概念浅谈Vuex 是一个专为 Vue.js 应用程序开发的状态管理模式。它采用集中式存储
- tf.nn.bidirectional_dynamic_rnn()函数：def bidirectional_dynamic_rnn( &nb
- 如何在聊天室实现趣味答题并计分功能？这个创意确实很好，我们可用在聊天室框架中加入一隐含帧（5分钟刷新一次）的做法来实现这一功能。questi
- 高层的期望“3个月内，我希望网站能增加X注册用户，每日的独立IP到Y，网站盈利达到Z……”作为一个团队的领袖或者产品负责人，这样的期望是根据
- 进度条是当我们处理冗长的任务时使用的控件，它是以动画的形式让用户知道该任务正在取得进展。在PyQt5中的进度条对应组件是QProgressB
- 前面章节我们介绍了如何选择优化的数据类型、如何高效的使用索引，这些对于高性能的MySQL来说是必不可少的。 但这些还完全不够，还需要合理的设
- 这是我的第一个真正意思上的自动化脚本。1、练习的测试用例为：打开百度首页，搜索“胡歌”，然后检索列表，有无“胡歌的新浪微博”这个链接 2、在
- 本文实例讲述了Python实现的企业粉丝抽奖功能。分享给大家供大家参考，具体如下：一 代码def scode9(schoice): &nbs
- 写过一篇"正则表达式30分钟入门教程"，有读者问：[^abc]表示不包含a、b、c中任意字符, 我想实现不包含字符串ab
- 管理认证系统最简单的方法是通过管理界面。然而，当你需要绝对的控制权的时候，有一些低层 API 需要深入专研，我们将在下面的章节中讨论它们。创
- 执行python脚本的时候，有时需要获取命令行参数的相关信息。C语言通过argc和argv来获取参数的个数和参数的内容，python中通过s
- Dim iSet conn=Server.CreateObject("ADODB.Connecti
- 先使用vue cli构建一个自己的vue项目1.npm i -g vue-cli2.vue init webpack sell (sell是
- 本文实例讲述了Python排序搜索基本算法之选择排序。分享给大家供大家参考，具体如下：选择排序就是第n次把序列中最小的元素排在第n的位置上，
- 本文实例为大家分享JavaScript弹出拖拽窗口的具体实现代码，供大家参考，具体内容如下需求说明： 1、点击页面按钮，弹出窗口；
