ThreadPoolExecutor線程池和ProcessPoolExecutor進程池

ProcessPoolExecutor線程池

1、為什么需要線程池呢，如果創建了20個線程，而同時只允許3個線程在運行，但是20個線程都需要創建和銷毀，線程的創建是需要消耗系統資源的，所以線程池的思想就是：每個線程各分配一個任務，剩下的任務皮隊等待，當某個線程完成了任務的時候，排隊任務就可以安排給這個線程繼續執行

2、標准庫concurrent.futures模塊，它提供了 ProcessPoolExecutor和ThreadPoolExecutor兩個類，實現了對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 = executor.submit(get_html,(3))
task2 = executor.submit(get_html,(2))
#done方法用於判斷某個任務是否完成
print(task1.done())
#cancel方法用於取消某個任務，該任務沒有放到線程池中才能被取消
print(task2.cancel())
print(task1.done())
#result方法可以獲取task的執行結果
print(task1.result())

#結果：
# get page 3s finished
# get page 2s finished
# True
# False
# True
# 3

• ThreadPoolExecutor構造實例的時候，傳入max_workers參數來設置線程中最多能同時運行的線程數目
• 使用submit函數來提交線程需要執行任務(函數名和參數)到線程池中，並返回該任務的句柄(類似於文件、畫圖)，注意submit()不是阻塞的，而是立即返回。
• 通過submit函數返回的任務句柄,能夠使用done()方法判斷該任務是否結束
• 使用result()方法可以獲取任務的返回值，查看內部代碼，發現這個方法是阻塞的

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
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
all_task = [executor.submit(get_html,(url)) for 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

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
executor = ThreadPoolExecutor(max_workers=2)
urls = [3,2,4]
for data in executor.map(get_html,urls):
    print("in main:get page {}s success".format(data))

#結果
# 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方法可以讓主線程阻塞,直到滿足設定的要求。wait方法接收3個參數，等待的任務序列、超時時間以及等待條件。等待條件return_when默認為ALL_COMPLETED,表明要等待所有的任務都借宿。可以看到運行結果中，確實是所有任務都完成了，主線程才打印出main,等待條件還可以設置為FIRST_COMPLETED,表示第一個任務完成就停止等待

# -*-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
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

ProcessPoolExecutor

方式一、同步調用方式:提交任務，原地等待任務執行結束，拿到任務返回結果。再執行下一行代碼會導致任務串行執行

# -*- coding:utf-8 -*-
# 方式一、同步調用方式:提交任務，原地等待任務執行結束，拿到任務返回結果。再執行下一行代碼會導致任務串行執行

# 進程池的兩種任務提交方式

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("主")

異步提交方式

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('主')

進程池的同步調用方式

同步調用優點解耦合,缺點:速度慢

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

異步調用方式

異步調用方式：缺點存在耦合，優點速度快

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

2-1-4進程池，異步調用+回調函數:解決耦合，速度慢,(異步調用＋回調)

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

線程池:異步+回調 ----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

1、線程不是越多越好，會涉及cpu上下文的切換（會把上一次的記錄保存）。

2、進程比線程消耗資源，進程相當於一個工廠，工廠里有很多人，里面的人共同享受着福利資源，，一個進程里默認只有一個主線程，比如：開啟程序是進程，里面執行的是線程，線程只是一個進程創建多個人同時去工作。

3、線程里有GIL全局解鎖器：不允許cpu調度

4、計算密度型適用於多進程

5、線程：線程是計算機中工作的最小單元

6、進程：默認有主線程 (幫工作)可以多線程共存

7、協程：一個線程，一個進程做多個任務,使用進程中一個線程去做多個任務，微線程

8、GIL全局解釋器鎖：保證同一時刻只有一個線程被cpu調度

來自：https://www.jianshu.com/p/b9b3d66aa0be

來自：https://blog.csdn.net/qq_33961117/article/details/82587873