python中ThreadPoolExecutor執行緒池和ProcessPoolExecutor程序池

2022-06-16 18:01:57

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排程

到此這篇關於python中ThreadPoolExecutor執行緒池和ProcessPoolExecutor程序池的文章就介紹到這了,更多相關python ThreadPoolExecutor 內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！