python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池

更新時間：2022年06月16日 16:05:42 作者：??HZ在掘金????

這篇文章主要介紹了python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池，文章圍繞主題相關(guān)資料展開詳細(xì)的內(nèi)容介紹，具有一定的參考價值，感興趣的小伙伴可以參考一下

1、ThreadPoolExecutor多線程

<1>為什么需要線程池呢?

對于io密集型，提高執(zhí)行的效率。
線程的創(chuàng)建是需要消耗系統(tǒng)資源的。

所以線程池的思想就是：每個線程各自分配一個任務(wù)，剩下的任務(wù)排隊(duì)等待，當(dāng)某個線程完成了任務(wù)的時候，排隊(duì)任務(wù)就可以安排給這個線程繼續(xù)執(zhí)行。

<2>標(biāo)準(zhǔn)庫concurrent.futures模塊

它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個類，
分別實(shí)現(xiàn)了對threading模塊和multiprocessing模塊的進(jìn)一步抽象。

不僅可以幫我們自動調(diào)度線程，還可以做到:

主線程可以獲取某一個線程(或者任務(wù))的狀態(tài)，以及返回值
當(dāng)一個線程完成的時候，主線程能夠立即知道
讓多線程和多進(jìn)程的編碼接口一致

<3>簡單使用

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor
import time

# 參數(shù)times用來模擬網(wǎng)絡(luò)請求時間
def get_html(times):
    print("get page {}s finished".format(times))
   return times
# 創(chuàng)建線程池
# 設(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()方法用于判斷某個任務(wù)是否完成，bool型，完成返回True，沒有完成返回False
print( task1.done() )
# cancel()方法用于取消某個任務(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
# 3

ThreadPoolExecutor類在構(gòu)造實(shí)例的時候，傳入max_workers參數(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é)束的方法，但是不能在主線程中一直判斷，有時候我們是得知某個任務(wù)結(jié)束了，就去獲取結(jié)果，而不是一直判斷每個任務(wù)有沒有結(jié)束。這時候就可以使用as_completed方法一次取出所有任務(wù)的結(jié)果。

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

# 參數(shù)times用來模擬網(wǎng)絡(luò)請求時間
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times

# 創(chuàng)建線程池子
# 設(shè)置最多2個線程運(yùn)行，其他等待
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
# 一次性把所有的任務(wù)都放進(jìn)線程池，得到一個句柄，但是最多只能同時執(zhí)行2個任務(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é)果可以看到，并不是先傳入哪個url，就先執(zhí)行哪個url，沒有先后順序

<5>map()方法

除了上面的as_completed()方法，還可以使用execumap方法。但是有一點(diǎn)不同，使用map方法，不需提前使用submit方法，
map方法與python標(biāo)準(zhǔn)庫中的map含義相同，都是將序列中的每個元素都執(zhí)行同一個函數(shù)。上面的代碼就是對urls列表中的每個元素都執(zhí)行g(shù)et_html()函數(shù)，并分配各線程池?？梢钥吹綀?zhí)行結(jié)果與上面的as_completed方法的結(jié)果不同，輸出順序和urls列表的順序相同，就算2s的任務(wù)先執(zhí)行完成，也會先打印出3s的任務(wù)先完成，再打印2s的任務(wù)完成

# -*-coding:utf-8 -*-
from concurrent.futures import ThreadPoolExecutor,as_completed
import time
# 參數(shù)times用來模擬網(wǎng)絡(luò)請求時間
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
# 創(chuàng)建線程池子
# 設(shè)置最多2個線程運(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個參數(shù)，等待的任務(wù)序列、超時時間以及等待條件。
等待條件return_when默認(rèn)為ALL_COMPLETED,表明要等待所有的任務(wù)都借宿?？梢钥吹竭\(yùn)行結(jié)果中，確實(shí)是所有任務(wù)都完成了，主線程才打印出main，等待條件還可以設(shè)置為FIRST_COMPLETED,表示第一個任務(wù)完成就停止等待。

超時時間參數(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ò)請求時間
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
   
# 創(chuàng)建線程池子
# 設(shè)置最多2個線程運(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
# main

2、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("主")
################
################
# 另一個同步調(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('完成時間:',time.time()-start)
    #完成時間: 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('主')
##################
##################
# 另一個異步調(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("完成時間",time.time()-start)#  完成時間 6.293345212936401

<3>怎么使用異步調(diào)用方式，但同時避免耦合的問題？

(1)進(jìn)程池：異步 + 回調(diào)函數(shù)，，cpu密集型，同時執(zhí)行，每個進(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)
    # 傳入的是個對象，獲取返回值 需要進(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會使用future對象的返回值，對象返回值是執(zhí)行任務(wù)的返回值
        #回調(diào)應(yīng)該是相當(dāng)于parse(future)
        future.add_done_callback(parse)
   p.shutdown(wait=True)
    print("完成時間",time.time()-start)#完成時間 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("完成時間",time.time()-start)#完成時間 32.52604126930237

3、總結(jié)

1、線程不是越多越好，會涉及cpu上下文的切換（會把上一次的記錄保存）。
2、進(jìn)程比線程消耗資源，進(jìn)程相當(dāng)于一個工廠，工廠里有很多人，里面的人共同享受著福利資源，，一個進(jìn)程里默認(rèn)只有一個主線程，比如：開啟程序是進(jìn)程，里面執(zhí)行的是線程，線程只是一個進(jìn)程創(chuàng)建多個人同時去工作。
3、線程里有GIL全局解鎖器：不允許cpu調(diào)度
4、計(jì)算密度型適用于多進(jìn)程
5、線程：線程是計(jì)算機(jī)中工作的最小單元
6、進(jìn)程：默認(rèn)有主線程 (幫工作)可以多線程共存
7、協(xié)程：一個線程，一個進(jìn)程做多個任務(wù),使用進(jìn)程中一個線程去做多個任務(wù)，微線程
8、GIL全局解釋器鎖：保證同一時刻只有一個線程被cpu調(diào)度

到此這篇關(guān)于python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池的文章就介紹到這了,更多相關(guān)python ThreadPoolExecutor 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

亚洲乱码中文字幕综合,中国熟女仑乱hd,亚洲精品乱拍国产一区二区三区,一本大道卡一卡二卡三乱码全集资源,又粗又黄又硬又爽的免费视频

軟件下載

源碼下載

軟件編程

網(wǎng)絡(luò)編程

在線工具

數(shù)據(jù)庫

CMS

常用工具

python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池

目錄

1、ThreadPoolExecutor多線程

<1>為什么需要線程池呢?

<2>標(biāo)準(zhǔn)庫concurrent.futures模塊

<3>簡單使用

<4>as_completed（一次性獲取所有的結(jié)果）

<5>map()方法

<6>wait()方法

2、ProcessPoolExecutor多進(jìn)程

<1>同步調(diào)用方式: 調(diào)用，然后等返回值，能解耦，但是速度慢

<2>異步調(diào)用方式：只調(diào)用，不等返回值，可能存在耦合，但是速度快

<3>怎么使用異步調(diào)用方式，但同時避免耦合的問題？

3、總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池

目錄

1、ThreadPoolExecutor多線程

<1>為什么需要線程池呢?

<2>標(biāo)準(zhǔn)庫concurrent.futures模塊

<3>簡單使用

<4>as_completed（一次性獲取所有的結(jié)果）

<5>map()方法

<6>wait()方法

2、ProcessPoolExecutor多進(jìn)程

<1>同步調(diào)用方式: 調(diào)用，然后等返回值，能解耦，但是速度慢

<2>異步調(diào)用方式：只調(diào)用，不等返回值，可能存在耦合，但是速度快

<3>怎么使用異步調(diào)用方式，但同時避免耦合的問題？

3、總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

1、ThreadPoolExecutor多線程

2、ProcessPoolExecutor多進(jìn)程

<1>同步調(diào)用方式: 調(diào)用，然后等返回值，能解耦，但是速度慢

<2>異步調(diào)用方式：只調(diào)用，不等返回值，可能存在耦合，但是速度快

<3>怎么使用異步調(diào)用方式，但同時避免耦合的問題？