使用concurrent.futures模塊中的線程池與進程池

線程池與進程池

以線程池舉例，系統使用多線程方式運行時，會產生大量的線程創建與銷毀，創建與銷毀必定會帶來一定的消耗，甚至導致系統資源的崩潰，這時使用線程池就是一個很好的解決方式。

“池”就說明瞭這裡邊維護了不止一個線程，線程池會提前創建好規定數量的線程，把需要使用多線程的任務提交給線程池，線程池會自己選擇空閑的線程來執行提交的任務，任務完成後，線程並不會在池子中銷毀，而是繼續存在並等待完成下一個分配的任務。當線程池以滿的時候，提交的線程會等待，也就是說線程池會有一個最大數量的運行線程限制。

進程池同樣也是這個道理。

concurrent.futures模塊為我們提供了ThreadPoolExecutor與ProcessPoolExecutor來使用線程進程池

ThreadPoolExecutor

下麵是一個簡單的例子

from concurrent.futures import ThreadPoolExecutor
import requests,time
url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
def get_url(url):
    content = requests.get(url).content.decode()
    print(url+'已獲取')

pool = ThreadPoolExecutor(max_workers=3)

start = time.time()
for url in url_list:
    future = pool.submit(get_url,url)
    ＃　print(future)
end = time.time()
print(end-start)

輸出的結果為：

0.0016434192657470703
https://www.cnblogs.com/已獲取
https://www.csdn.net/已獲取
https://github.com/已獲取

例子中max_workers為指定線程個數，pool.submit為提交任務到線程執行，get_url為方法，url為參數

並且通過輸出順序可以看到線程池的執行並不會阻塞主線程的運行

print(future)被打了註釋，現在我們取消註釋運行一下：

Future at 0x7ff6cfaa8860 state=running
Future at 0x7ff6ce965860 state=running
Future at 0x7ff6ce96e278 state=running
0.006175518035888672
https://www.cnblogs.com/已獲取
https://www.csdn.net/已獲取
https://github.com/已獲取

每提交一個任務後都會返回一個future對象，通過它可以查看任務運行的狀態，state=running表示正在運行

future對象還有許多方法：

future.done()

from concurrent.futures import ThreadPoolExecutor
import requests,time
url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
def get_url(url):
    content = requests.get(url).content.decode()
    print(url+'已獲取')

pool = ThreadPoolExecutor(max_workers=3)
future_list = []
start = time.time()
for url in url_list:
    future = pool.submit(get_url,url)
    print(future.done())
    future_list.append(future)
end = time.time()

print(end-start)
time.sleep(5)
for future in future_list:
    print(future.done())

這裡添加了future_list,為了顯示效果中間添加sleep,最後結果為：

False
False
False
0.001546621322631836
https://www.cnblogs.com/已獲取
https://www.csdn.net/已獲取
https://github.com/已獲取
True
True
True

future.done()可以顯示當前允許狀態

future.result()

from concurrent.futures import ThreadPoolExecutor
import requests,time
url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
def get_url(url):
    content = requests.get(url).content.decode()
    print(url+'已獲取')
    return url

pool = ThreadPoolExecutor(max_workers=3)
future_list = []
start = time.time()
for url in url_list:
    future = pool.submit(get_url,url)
    print(future.result())
    future_list.append(future)
end = time.time()

print(end-start)
for future in future_list:
    print(future.result())

結果為：

https://www.cnblogs.com/已獲取
https://www.cnblogs.com/
https://www.csdn.net/已獲取
https://www.csdn.net/
https://github.com/已獲取
https://github.com/
2.0975613594055176
https://www.cnblogs.com/
https://www.csdn.net/
https://github.com/

可見result()方法可以得到任務的返回值，但會阻塞，因為不運行完怎麼會得到返回值呢？

除此之外還有很多方法：

使用map方法

from concurrent.futures import ThreadPoolExecutor
import requests,time
url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
def get_url(url):
    content = requests.get(url).content.decode()
    print(url+'已獲取')
    return url

pool = ThreadPoolExecutor(max_workers=3)

pool.map(get_url,url_list)

與內建函數用法類似

使用wait方法

from concurrent.futures import ThreadPoolExecutor,wait
import requests,time
url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
def get_url(url):
    content = requests.get(url).content.decode()
    print(url+'已獲取')
    return url

pool = ThreadPoolExecutor(max_workers=3)
future_list = []
start = time.time()
for url in url_list:
    future = pool.submit(get_url,url)
    future_list.append(future)

print(wait(future_list))
end = time.time()
print(end-start)

https://www.cnblogs.com/已獲取
https://www.csdn.net/已獲取
https://github.com/已獲取
DoneAndNotDoneFutures(done={Future at 0x7f7506447da0 state=finished returned str, Future at 0x7f75074c9828 state=finished returned str, Future at 0x7f75064477f0 state=finished returned str}, not_done=set())

6.678021430969238

wait返回值是一個元組，元組裡是已完成和未完成的兩個集合，它的return_when參數接受３個選項FIRST_COMPLETED, FIRST_EXCEPTION 和ALL_COMPLETE，預設是ALL_COMPLETE，意味著所有都完成，FIRST_COMPLETED意味著有一個完成了就可以了， FIRST_EXCEPTION是第一個出現異常就會停止wait

例如：

from concurrent.futures import ThreadPoolExecutor,wait
import requests,time
url_list = ['https://www.cnblogs.com/', 'https://www.csdn.net/', 'https://github.com/']
def get_url(url):
    content = requests.get(url).content.decode()
    print(url+'已獲取')
    return url

def error(url):
    gg

pool = ThreadPoolExecutor(max_workers=4)
future_list = []
start = time.time()
future_list.append(pool.submit(error,'https://www.cnblogs.com/'))
for url in url_list:
    future = pool.submit(get_url,url)
    future_list.append(future)

print(wait(future_list,return_when='FIRST_EXCEPTION'))
end = time.time()
print(end-start)

DoneAndNotDoneFutures(done={Future at 0x7fd1a5b95320 state=finished raised NameError}, not_done={Future at 0x7fd1a4b11a90 state=running, Future at 0x7fd1a4b11a20 state=running, Future at 0x7fd1a4c897f0 state=running})
0.001996755599975586
https://www.cnblogs.com/已獲取
https://www.csdn.net/已獲取
https://github.com/已獲取

ProcessPoolExecutor

進程池與線程池的使用方式基本相同，套用即可