python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池
1、ThreadPoolExecutor多線程
<1>為什么需要線程池呢?
- 對(duì)于io密集型,提高執(zhí)行的效率。
- 線程的創(chuàng)建是需要消耗系統(tǒng)資源的。
所以線程池的思想就是:每個(gè)線程各自分配一個(gè)任務(wù),剩下的任務(wù)排隊(duì)等待,當(dāng)某個(gè)線程完成了任務(wù)的時(shí)候,排隊(duì)任務(wù)就可以安排給這個(gè)線程繼續(xù)執(zhí)行。
<2>標(biāo)準(zhǔn)庫concurrent.futures模塊
它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個(gè)類,
分別實(shí)現(xiàn)了對(duì)threading模塊和multiprocessing模塊的進(jìn)一步抽象。
不僅可以幫我們自動(dòng)調(diào)度線程,還可以做到:
- 主線程可以獲取某一個(gè)線程(或者任務(wù))的狀態(tài),以及返回值
- 當(dāng)一個(gè)線程完成的時(shí)候,主線程能夠立即知道
- 讓多線程和多進(jìn)程的編碼接口一致
<3>簡(jiǎn)單使用
# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor
import time
# 參數(shù)times用來模擬網(wǎng)絡(luò)請(qǐng)求時(shí)間
def get_html(times):
print("get page {}s finished".format(times))
return times
# 創(chuàng)建線程池
# 設(shè)置線程池中最多能同時(shí)運(yùn)行的線程數(shù)目,其他等待
executor = ThreadPoolExecutor(max_workers=2)
# 通過submit函數(shù)提交執(zhí)行的函數(shù)到線程池中,submit函數(shù)立即返回,不阻塞
# task1和task2是任務(wù)句柄
task1 = executor.submit( get_html, (2) )
task2 = executor.submit( get_html, (3) )
# done()方法用于判斷某個(gè)任務(wù)是否完成,bool型,完成返回True,沒有完成返回False
print( task1.done() )
# cancel()方法用于取消某個(gè)任務(wù),該任務(wù)沒有放到線程池中才能被取消,如果已經(jīng)放進(jìn)線程池子中,則不能被取消
# bool型,成功取消了返回True,沒有取消返回False
print( task2.cancel() )
# result()方法可以獲取task的執(zhí)行結(jié)果,前提是get_html()函數(shù)有返回值
print( task1.result() )
print( task2.result() )
# 結(jié)果:
# get page 3s finished
# get page 2s finished
# True
# False
# 2
# 3ThreadPoolExecutor類在構(gòu)造實(shí)例的時(shí)候,傳入max_workers參數(shù)來設(shè)置線程池中最多能同時(shí)運(yùn)行的線程數(shù)目
使用submit()函數(shù)來提交線程需要執(zhí)行任務(wù)(函數(shù)名和參數(shù))到線程池中,并返回該任務(wù)的句柄,
注意:submit()不是阻塞的,而是立即返回。
通過submit()函數(shù)返回的任務(wù)句柄,能夠使用done()方法判斷該任務(wù)是否結(jié)束,使用cancel()方法來取消,使用result()方法可以獲取任務(wù)的返回值,查看內(nèi)部代碼,發(fā)現(xiàn)該方法是阻塞的
<4>as_completed(一次性獲取所有的結(jié)果)
上面雖然提供了判斷任務(wù)是否結(jié)束的方法,但是不能在主線程中一直判斷,有時(shí)候我們是得知某個(gè)任務(wù)結(jié)束了,就去獲取結(jié)果,而不是一直判斷每個(gè)任務(wù)有沒有結(jié)束。這時(shí)候就可以使用as_completed方法一次取出所有任務(wù)的結(jié)果。
# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
# 參數(shù)times用來模擬網(wǎng)絡(luò)請(qǐng)求時(shí)間
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times
# 創(chuàng)建線程池子
# 設(shè)置最多2個(gè)線程運(yùn)行,其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
# 一次性把所有的任務(wù)都放進(jìn)線程池,得到一個(gè)句柄,但是最多只能同時(shí)執(zhí)行2個(gè)任務(wù)
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))
# 結(jié)果
# 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
# 從結(jié)果可以看到,并不是先傳入哪個(gè)url,就先執(zhí)行哪個(gè)url,沒有先后順序<5>map()方法
除了上面的as_completed()方法,還可以使用execumap方法。但是有一點(diǎn)不同,使用map方法,不需提前使用submit方法,
map方法與python標(biāo)準(zhǔn)庫中的map含義相同,都是將序列中的每個(gè)元素都執(zhí)行同一個(gè)函數(shù)。上面的代碼就是對(duì)urls列表中的每個(gè)元素都執(zhí)行g(shù)et_html()函數(shù),并分配各線程池。可以看到執(zhí)行結(jié)果與上面的as_completed方法的結(jié)果不同,輸出順序和urls列表的順序相同,就算2s的任務(wù)先執(zhí)行完成,也會(huì)先打印出3s的任務(wù)先完成,再打印2s的任務(wù)完成
# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,as_completed
import time
# 參數(shù)times用來模擬網(wǎng)絡(luò)請(qǐng)求時(shí)間
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times
# 創(chuàng)建線程池子
# 設(shè)置最多2個(gè)線程運(yùn)行,其他等待
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))
結(jié)果:
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方法可以讓主線程阻塞,直到滿足設(shè)定的要求。wait方法接收3個(gè)參數(shù),等待的任務(wù)序列、超時(shí)時(shí)間以及等待條件。
等待條件return_when默認(rèn)為ALL_COMPLETED,表明要等待所有的任務(wù)都借宿??梢钥吹竭\(yùn)行結(jié)果中,確實(shí)是所有任務(wù)都完成了,主線程才打印出main,等待條件還可以設(shè)置為FIRST_COMPLETED,表示第一個(gè)任務(wù)完成就停止等待。
超時(shí)時(shí)間參數(shù)可以不設(shè)置:
wait()方法和as_completed(), map()沒有關(guān)系。不管你是用as_completed(),還是用map()方法,你都可以在執(zhí)行主線程之前使用wait()。
as_completed()和map()是二選一的。
# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED,FIRST_COMPLETED
import time
# 參數(shù)times用來模擬網(wǎng)絡(luò)請(qǐng)求時(shí)間
def get_html(times):
time.sleep(times)
print("get page {}s finished".format(times))
return times
# 創(chuàng)建線程池子
# 設(shè)置最多2個(gè)線程運(yùn)行,其他等待
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")
# 結(jié)果
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main2、ProcessPoolExecutor多進(jìn)程
<1>同步調(diào)用方式: 調(diào)用,然后等返回值,能解耦,但是速度慢
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) # 設(shè)置
for i in range(10):
# 同步調(diào)用方式,不僅要調(diào)用,還要等返回值
obj = p.submit(task, "進(jìn)程pid:") # 傳參方式(任務(wù)名,參數(shù)),參數(shù)使用位置或者關(guān)鍵字參數(shù)
res = obj.result()
p.shutdown(wait=True) # 關(guān)閉進(jìn)程池的入口,等待池內(nèi)任務(wù)運(yùn)行結(jié)束
print("主")
################
################
# 另一個(gè)同步調(diào)用的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 解析結(jié)果為%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) # 需要等結(jié)果,所以是同步調(diào)用
l.append(future)
# 關(guān)閉進(jìn)程池,等所有的進(jìn)程執(zhí)行完畢
p.shutdown(wait=True)
for future in l:
parse(future.result())
print('完成時(shí)間:',time.time()-start)
#完成時(shí)間: 13.209137678146362<2>異步調(diào)用方式:只調(diào)用,不等返回值,可能存在耦合,但是速度快
def task(name):
print("%s %s is running" %(name,os.getpid()))
time.sleep(random.randint(1,3))
if __name__ == '__main__':
p = ProcessPoolExecutor(4) # 設(shè)置進(jìn)程池內(nèi)進(jìn)程
for i in range(10):
# 異步調(diào)用方式,只調(diào)用,不等返回值
p.submit(task,'進(jìn)程pid:') # 傳參方式(任務(wù)名,參數(shù)),參數(shù)使用位置參數(shù)或者關(guān)鍵字參數(shù)
p.shutdown(wait=True) # 關(guān)閉進(jìn)程池的入口,等待池內(nèi)任務(wù)運(yùn)行結(jié)束
print('主')
##################
##################
# 另一個(gè)異步調(diào)用的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 解析結(jié)果為%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("完成時(shí)間",time.time()-start)# 完成時(shí)間 6.293345212936401<3>怎么使用異步調(diào)用方式,但同時(shí)避免耦合的問題?
(1)進(jìn)程池:異步 + 回調(diào)函數(shù),,cpu密集型,同時(shí)執(zhí)行,每個(gè)進(jìn)程有不同的解釋器和內(nèi)存空間,互不干擾
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)
# 傳入的是個(gè)對(duì)象,獲取返回值 需要進(jìn)行result操作
res = future.result()
print("res",)
print('%s 解析結(jié)果為%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)
#模塊內(nèi)的回調(diào)函數(shù)方法,parse會(huì)使用future對(duì)象的返回值,對(duì)象返回值是執(zhí)行任務(wù)的返回值
#回調(diào)應(yīng)該是相當(dāng)于parse(future)
future.add_done_callback(parse)
p.shutdown(wait=True)
print("完成時(shí)間",time.time()-start)#完成時(shí)間 33.79998469352722(2)線程池:異步 + 回調(diào)函數(shù),IO密集型主要使用方式,線程池:執(zhí)行操作為誰有空誰執(zhí)行
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 解析結(jié)果為%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("完成時(shí)間",time.time()-start)#完成時(shí)間 32.526041269302373、總結(jié)
- 1、線程不是越多越好,會(huì)涉及cpu上下文的切換(會(huì)把上一次的記錄保存)。
- 2、進(jìn)程比線程消耗資源,進(jìn)程相當(dāng)于一個(gè)工廠,工廠里有很多人,里面的人共同享受著福利資源,,一個(gè)進(jìn)程里默認(rèn)只有一個(gè)主線程,比如:開啟程序是進(jìn)程,里面執(zhí)行的是線程,線程只是一個(gè)進(jìn)程創(chuàng)建多個(gè)人同時(shí)去工作。
- 3、線程里有GIL全局解鎖器:不允許cpu調(diào)度
- 4、計(jì)算密度型適用于多進(jìn)程
- 5、線程:線程是計(jì)算機(jī)中工作的最小單元
- 6、進(jìn)程:默認(rèn)有主線程 (幫工作)可以多線程共存
- 7、協(xié)程:一個(gè)線程,一個(gè)進(jìn)程做多個(gè)任務(wù),使用進(jìn)程中一個(gè)線程去做多個(gè)任務(wù),微線程
- 8、GIL全局解釋器鎖:保證同一時(shí)刻只有一個(gè)線程被cpu調(diào)度
到此這篇關(guān)于python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池的文章就介紹到這了,更多相關(guān)python ThreadPoolExecutor 內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
上手簡(jiǎn)單,功能強(qiáng)大的Python爬蟲框架——feapder
這篇文章主要介紹了上手簡(jiǎn)單,功能強(qiáng)大的Python爬蟲框架——feapder的使用教程,幫助大家更好的利用python進(jìn)行爬蟲,感興趣的朋友可以了解下2021-04-04
python實(shí)現(xiàn)輸出一個(gè)序列的所有子序列示例
今天小編就為大家分享一篇python實(shí)現(xiàn)輸出一個(gè)序列的所有子序列示例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2019-11-11
Django項(xiàng)目開發(fā)中cookies和session的常用操作分析
這篇文章主要介紹了Django項(xiàng)目開發(fā)中cookies和session的常用操作,結(jié)合實(shí)例形式分析了Django中cookie與session的檢查、設(shè)置、獲取等常用操作技巧,需要的朋友可以參考下2018-07-07
簡(jiǎn)單快捷:NumPy入門教程的環(huán)境設(shè)置
NumPy是Python語言的一個(gè)擴(kuò)展程序庫,支持高階大量的維度數(shù)組與矩陣運(yùn)算,此外也針對(duì)數(shù)組運(yùn)算提供大量的數(shù)學(xué)函數(shù)庫,本教程是為那些想了解NumPy的基礎(chǔ)知識(shí)和各種功能的人準(zhǔn)備的,它對(duì)算法開發(fā)人員特別有用,需要的朋友可以參考下2023-10-10
python Flask實(shí)現(xiàn)restful api service
本篇文章主要介紹了python Flask實(shí)現(xiàn)restful api service,小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過來看看吧2017-12-12
在Python3.74+PyCharm2020.1 x64中安裝使用Kivy的詳細(xì)教程
這篇文章主要介紹了在Python3.74+PyCharm2020.1 x64中安裝使用Kivy的詳細(xì)教程,本文通過圖文實(shí)例相結(jié)合給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-08-08
python3中類的繼承以及self和super的區(qū)別詳解
今天小編就為大家分享一篇python3中類的繼承以及self和super的區(qū)別詳解,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2019-06-06

