溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹了python中ThreadPoolExecutor線程池和ProcessPoolExecutor進程池怎么使用的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇python中ThreadPoolExecutor線程池和ProcessPoolExecutor進程池怎么使用文章都會有所收獲，下面我們一起來看看吧。
對于io密集型，提高執行的效率。
線程的創建是需要消耗系統資源的。
所以線程池的思想就是：每個線程各自分配一個任務，剩下的任務排隊等待，當某個線程完成了任務的時候，排隊任務就可以安排給這個線程繼續執行。
它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個類，
分別實現了對threading模塊和multiprocessing模塊的進一步抽象。
不僅可以幫我們自動調度線程，還可以做到:
主線程可以獲取某一個線程(或者任務)的狀態，以及返回值
當一個線程完成的時候，主線程能夠立即知道
讓多線程和多進程的編碼接口一致
# -*-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
# 3ThreadPoolExecutor類在構造實例的時候，傳入max_workers參數來設置線程池中最多能同時運行的線程數目
使用submit()函數來提交線程需要執行任務(函數名和參數)到線程池中，并返回該任務的句柄，
注意：submit()不是阻塞的，而是立即返回。
通過submit()函數返回的任務句柄，能夠使用done()方法判斷該任務是否結束，使用cancel()方法來取消，使用result()方法可以獲取任務的返回值，查看內部代碼，發現該方法是阻塞的
上面雖然提供了判斷任務是否結束的方法，但是不能在主線程中一直判斷，有時候我們是得知某個任務結束了，就去獲取結果，而不是一直判斷每個任務有沒有結束。這時候就可以使用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，沒有先后順序除了上面的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
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
# mainimport 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.209137678146362def 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(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關于“python中ThreadPoolExecutor線程池和ProcessPoolExecutor進程池怎么使用”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“python中ThreadPoolExecutor線程池和ProcessPoolExecutor進程池怎么使用”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
