摘要：標(biāo)準(zhǔn)庫中所有阻塞型函數(shù)都會釋放，允許其他線程運(yùn)行。如果調(diào)用引發(fā)異常，那么當(dāng)從迭代器檢索其值時，將引發(fā)異常?？偨Y(jié)自版就支持線程了，只不過是使用線程的最新方式。類封裝了模塊的組件，使使用線程變得更加方便。下一篇筆記應(yīng)該是使用處理并發(fā)。

作為Python程序員，平時很少使用并發(fā)編程，偶爾使用也只需要派生出一批獨立的線程，然后放到隊列中，批量執(zhí)行。所以，不夸張的說，雖然我知道線程、進(jìn)程、并行、并發(fā)的概念，但每次使用的時候可能還需要再打開文檔回顧一下。

現(xiàn)在這一篇還是 《流暢的python》讀書筆記，譯者在這里把future 翻譯為“期物”，我覺得不太合適，既然future不能找到一個合適的詞匯，暫時還是直接使用 future 吧。

future 是一種對象，表示異步執(zhí)行的操作。這個概念是 concurrent.futures模塊和asyncio包的基礎(chǔ)。

concurrent.futures 模塊是Python3.2 引入的，對于Python2x 版本，Python2.5 以上的版本可以安裝 futures 包來使用這個模塊。

從Python3.4起，標(biāo)準(zhǔn)庫中有兩個為Future的類：concurrent.futures.Future 和 asyncio.Future。這兩個類作用相同：兩個Future類的實例都表示可能已經(jīng)完成或未完成的延遲計算。

Future 封裝待完成的操作，可放入隊列，完成的狀態(tài)可以查詢，得到結(jié)果（或拋出異常）后可以獲取結(jié)果（或異常）。

我們知道，如果程序中包含I/O操作，程序會有很高的延遲，CPU會處于等待狀態(tài)，這時如果我們不使用并發(fā)會浪費很多時間。

我們先舉個例子：

下邊是有兩段代碼，主要功能都是從網(wǎng)上下載人口前20的國際的國旗：
第一段代碼(flagss.py)是依序下載：下載完一個圖片后保存到硬盤，然后請求下一張圖片；
第二段代碼(flagss_threadpool.py)使用 concurrent.futures 模塊，批量下載10張圖片。

運(yùn)行分別運(yùn)行兩段代碼3次，結(jié)果如下：

images.py 的結(jié)果如下

$ python flags.py
BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN 
20 flags downloaded in 6.18s

$ python flags.py
BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN 
20 flags downloaded in 5.67s

$ python flags.py
BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN 
20 flags downloaded in 6.55s

可以看到，依次下載10張圖片，平均需要6秒

flags_threadpool.py 的結(jié)果如下：

$ python flags_threadpool.py
NG EG VN BR JP FR DE CN TR BD PK MX PH US RU IN ET CD ID IR 
20 flags downloaded in 2.12s

$ python flags_threadpool.py
BR IN DE FR TR RU EG NG JP CN ID ET PK MX PH US IR CD VN BD 
20 flags downloaded in 2.23s

$ python flags_threadpool.py
CN BR DE ID NG RU TR IN MX US IR BD VN CD PH EG FR JP ET PK 
20 flags downloaded in 1.18s

使用 concurrent.futures 后，下載10張圖片平均需要2秒

通過上邊的結(jié)果我們發(fā)現(xiàn)使用 concurrent.futures 后，下載效率大幅提升。

下邊我們來看下這兩段代碼。

同步執(zhí)行的代碼flags.py：

#! -*- coding: utf-8 -*-

import os
import time
import sys

import requests  # <1>

POP20_CC = ("CN IN US ID BR PK NG BD RU JP "
            "MX PH VN ET EG DE IR TR CD FR").split()  # <2>

BASE_URL = "http://flupy.org/data/flags"  # <3>

DEST_DIR = "images/"  # <4>


# 保存圖片
def save_flag(img, filename):  # <5>
    path = os.path.join(DEST_DIR, filename)
    with open(path, "wb") as fp:
        fp.write(img)


# 下載圖片
def get_flag(cc):  # <6>
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    # 這里我們使用 requests 包，需要先通過pypi安裝
    resp = requests.get(url)
    return resp.content


# 顯示一個字符串，然后刷新sys.stdout,目的是在一行消息中看到進(jìn)度
def show(text):  # <7>
    print(text, end=" ")
    sys.stdout.flush()


def download_many(cc_list):  # <8>
    for cc in sorted(cc_list):  # <9>
        image = get_flag(cc)
        show(cc)
        save_flag(image, cc.lower() + ".gif")

    return len(cc_list)


def main(download_many):  # <10>
    t0 = time.time()
    count = download_many(POP20_CC)
    elapsed = time.time() - t0
    msg = "
{} flags downloaded in {:.2f}s"
    print(msg.format(count, elapsed))


if __name__ == "__main__":
    main(download_many)  # <11>

使用 concurrent.future 并發(fā)的代碼 flags_threadpool.py

#! -*- coding: utf-8 -*-

from concurrent import futures

from flags import save_flag, get_flag, show, main

# 設(shè)定ThreadPoolExecutor 類最多使用幾個線程
MAX_WORKERS = 20


# 下載一個圖片
def download_one(cc):
    image = get_flag(cc)
    show(cc)
    save_flag(image, cc.lower() + ".gif")
    return cc


def download_many(cc_list):
    # 設(shè)定工作的線程數(shù)量，使用約需的最大值與要處理的數(shù)量直接較小的那個值，以免創(chuàng)建多余的線程
    workers = min(MAX_WORKERS, len(cc_list))  # <4>
    # 使用工作的線程數(shù)實例化ThreadPoolExecutor類；
    # executor.__exit__方法會調(diào)用executor.shutdown(wait=True)方法，
    # 它會在所有線程都執(zhí)行完畢前阻塞線程
    with futures.ThreadPoolExecutor(workers) as executor:  # <5>
        # map 與內(nèi)置map方法類似，不過download_one 函數(shù)會在多個線程中并發(fā)調(diào)用；
        # map 方法返回一個生成器，因此可以迭代，
        # 迭代器的__next__方法調(diào)用各個Future 的 result 方法
        res = executor.map(download_one, sorted(cc_list))

    # 返回獲取的結(jié)果數(shù)量；如果有現(xiàn)成拋出異常，會在這里拋出
    # 這與隱式調(diào)用next() 函數(shù)從迭代器中獲取相應(yīng)的返回值一樣。
    return len(list(res))  # <7>
    return len(results)


if __name__ == "__main__":
    main(download_many)

上邊的代碼，我們對 concurrent.futures 的使用有了大致的了解。但 future 在哪里呢，我們并沒有看到。

Future 是 concurrent.futures 模塊和 asyncio 包的重要組件。從Python3.4起，標(biāo)準(zhǔn)庫中有兩個為Future的類：concurrent.futures.Future 和 asyncio.Future。這兩個Future作用相同。

Future 封裝待完成的操作，可放入隊列，完成的狀態(tài)可以查詢，得到結(jié)果（或拋出異常）后可以獲取結(jié)果（或異常）。
Future 表示終將發(fā)生的事情，而確定某件事情會發(fā)生的唯一方式是執(zhí)行的時間已經(jīng)排定。因此只有把某件事交給 concurrent.futures.Executor 子類處理時，才會創(chuàng)建 concurrent.futures.Future 實例。

例如，調(diào)用Executor.submit() 方法的參數(shù)是一個可調(diào)用的對象，調(diào)用這個方法后會為傳入的可調(diào)用對象排期，并返回一個Future。

Future 有三個重要的方法：

.done() 返回布爾值，表示Future 是否已經(jīng)執(zhí)行

.add_done_callback() 這個方法只有一個參數(shù)，類型是可調(diào)用對象，F(xiàn)uture運(yùn)行結(jié)束后會回調(diào)這個對象。

.result() 如果 Future 運(yùn)行結(jié)束后調(diào)用result(), 會返回可調(diào)用對象的結(jié)果或者拋出執(zhí)行可調(diào)用對象時拋出的異常，如果是 Future 沒有運(yùn)行結(jié)束時調(diào)用 f.result()方法，這時會阻塞調(diào)用方所在的線程，直到有結(jié)果返回。此時result 方法還可以接收 timeout 參數(shù)，如果在指定的時間內(nèi) Future 沒有運(yùn)行完畢，會拋出 TimeoutError 異常。

asyncio.Future.result 方法不支持設(shè)定超時時間，如果想獲取 Future 的結(jié)果，可以使用 yield from 結(jié)構(gòu)

為了加深對 Future 的理解，現(xiàn)在我們修改下 flags_threadpool.py download_many 函數(shù)。

def download_many(cc_list):
    cc_list = cc_list[:5]
    with futures.ThreadPoolExecutor(max_workers=3) as executor:
        to_do = []
        # 用于創(chuàng)建并排定 future
        for cc in sorted(cc_list):
            # submit 方法排定可調(diào)用對象的執(zhí)行時間然后返回一個future，表示這個待執(zhí)行的操作
            future = executor.submit(download_one, cc)
            to_do.append(future)
            msg = "Scheduled for {}: {}"
            print(msg.format(cc, future))
        
        results = []
        # 用于獲取future 結(jié)果
        # as_completed 接收一個future 列表，返回值是一個迭代器，在運(yùn)行結(jié)束后產(chǎn)出future
        for future in futures.as_completed(to_do):
            res = future.result()
            msg = "{} result: {!r}"
            print(msg.format(future, res))
            results.append(res)
    
    return len(results)

現(xiàn)在執(zhí)行代碼，運(yùn)行結(jié)果如下：

Scheduled for BR: 
Scheduled for CN: 
Scheduled for ID: 
Scheduled for IN: 
Scheduled for US: 
BR  result: "BR"
IN  result: "IN"
CN  result: "CN"
ID  result: "ID"
US  result: "US"

5 flags downloaded in 1.47s

從結(jié)果可以看到，future 的 repr() 方法會顯示狀態(tài)，前三個 是running 是因為我們設(shè)定了三個進(jìn)程，所以后兩個是pendding 狀態(tài)。如果將max_workers參數(shù)設(shè)置為5，結(jié)果就會全都是 running。

雖然，使用 future 的腳步比第一個腳本的執(zhí)行速度快了很多，但由于受GIL的限制，下載并不是并行的。

CPython 解釋器本身不是線程安全的，因此解釋器被一個全局解釋器鎖保護(hù)著，它確保任何時候都只有一個Python線程執(zhí)行。

然而，Python標(biāo)準(zhǔn)庫中所有執(zhí)行阻塞型I/O操作的函數(shù)，在等待系統(tǒng)返回結(jié)果時都會釋放GIL。這意味著I/O密集型Python程序能從中受益：一個Python線程等待網(wǎng)絡(luò)響應(yīng)時，阻塞型I/O函數(shù)會釋放GIL，再運(yùn)行一個線程。

Python 標(biāo)準(zhǔn)庫中所有阻塞型I/O函數(shù)都會釋放GIL，允許其他線程運(yùn)行。time.sleep()函數(shù)也會釋放GIL。

那么如何在CPU密集型作業(yè)中使用 concurrent.futures 模塊繞開GIL呢？

答案是 使用 ProcessPoolExecutor 類。

使用這個模塊可以在做CPU密集型工作是繞開GIL，利用所有可用核心。

ThreadPoolExecutor 和 ProcessPoolExecutor 都實現(xiàn)了通用的 Executor 接口，所以，我們可以輕松的將基于線程的方案改為使用進(jìn)程的方案。

比如下邊這樣：

def download_many(cc_list):
    workers = min(MAX_WORKERS, len(cc_list))
    with futures.ThreadPoolExecutor(workers) as executor:
        pass

# 改成
def download_many(cc_list):
    with futures.ProcessPoolExecutor() as executor:
        pass

需要注意的是，ThreadPoolExecutor 需要指定 max_workers 參數(shù)，
而 ProcessPoolExecutor 的這個參數(shù)是可選的默認(rèn)值是 os.cup_count()(計算機(jī)cpu核心數(shù))。

ProcessPoolExecutor 的價值主要體現(xiàn)在CPU密集型作業(yè)上。

使用Python處理CPU密集型工作，應(yīng)該試試PyPy，會有更高的執(zhí)行速度。

現(xiàn)在我們回到開始的代碼，看下 Executor.map 函數(shù)。

文檔中對map函數(shù)的介紹如下。

map(func, *iterables, timeout=None, chunksize=1)
等同于 map(func, *iterables)，不同的是 func 是異步執(zhí)行的，并且可以同時進(jìn)行對 func 的多個調(diào)用。如果調(diào)用 __next__()，則返回的迭代器提出 concurrent.futures.TimeoutError，并且在從 Executor.map() 的原始調(diào)用起的 timeout 秒之后結(jié)果不可用。 timeout 可以是int或float。如果未指定 timeout 或 None，則等待時間沒有限制。如果調(diào)用引發(fā)異常，那么當(dāng)從迭代器檢索其值時，將引發(fā)異常。當(dāng)使用 ProcessPoolExecutor 時，此方法將 iterables 分成多個塊，它作為多帶帶的任務(wù)提交到進(jìn)程池。這些塊的（近似）大小可以通過將 chunksize 設(shè)置為正整數(shù)來指定。對于非常長的迭代，與默認(rèn)大小1相比，使用大值 chunksize 可以顯著提高性能。使用 ThreadPoolExecutor，chunksize 沒有效果。
在 3.5 版更改: 添加了 chunksize 參數(shù)。

Executor.map 還有個特性比較有用，那就是這個函數(shù)返回結(jié)果的順序于調(diào)用開始的順序是一致的。如果第一個調(diào)用稱其結(jié)果用時10秒，其他調(diào)用只用1秒，代碼會阻塞10秒，獲取map方法返回的生成器產(chǎn)出的第一個結(jié)果。

如果不是獲取到所有結(jié)果再處理，通常會使用 Executor.submit + Executor.as_completed 組合使用的方案。

Executor.submit + Executor.as_completed 這個組合更靈活，因為submit方法能處理不同的可調(diào)用對象和參數(shù)，而executor.map 只能處理參數(shù)不同的同一個可調(diào)用對象。此外，傳給futures.as_completed 函數(shù)的期物集合可以來自不同的 Executor 實例。

futures 有三個異常類：

exception concurrent.futures.CancelledError 在future取消時引發(fā)。

exception concurrent.futures.TimeoutError 在future操作超過給定超時時觸發(fā)。

exception concurrent.futures.process.BrokenProcessPool
從 RuntimeError 派生，當(dāng) ProcessPoolExecutor 的一個工人以非干凈方式終止（例如，如果它從外部被殺死）時，引發(fā)此異常類。

我們先看一下，future.result() 出現(xiàn)異常的處理情況。代碼改動如下：

# 將第一個 CN 改為CN1 也可以是其它任意錯誤代碼
POP20_CC = ("CN1 IN US ID BR PK NG BD RU JP "
            "MX PH VN ET EG DE IR TR CD FR").split()


def get_flag(cc):  # <6>
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    resp = requests.get(url)
    if resp.status_code != 200:  # <1>
        resp.raise_for_status() # 如果不是200 拋出異常
    return resp.content

def download_one(cc):
    try:
        image = get_flag(cc)
    # 捕獲 requests.exceptions.HTTPError
    except requests.exceptions.HTTPError as exc:  #
        # 如果有異常 直接拋出
        raise
    else:
        save_flag(image, cc.lower() + ".gif")
    return cc

現(xiàn)在執(zhí)行代碼，會發(fā)現(xiàn) download_one 中的異常傳遞到了download_many 中,并且導(dǎo)致拋出了異常，未執(zhí)行完的其它future 也都中斷。

為了能保證其它沒有錯誤的future 可以正常執(zhí)行，這里我們需要對future.result() 做異常處理。

改動結(jié)果如下：

def download_many(cc_list):
    cc_list = cc_list[:5]
    with futures.ThreadPoolExecutor(max_workers=20) as executor:
        to_do_map = {}
        for cc in sorted(cc_list):
            future = executor.submit(download_one, cc)
            to_do_map[future] = cc
            msg = "Scheduled for {}: {}"
            print(msg.format(cc, future))

        results = []
        for future in futures.as_completed(to_do_map):
            try:
                res = future.result()
            except requests.exceptions.HTTPError as exc:
                # 處理可能出現(xiàn)的異常
                error_msg = "{} result {}".format(cc, exc)
            else:
                error_msg = ""
            if error_msg:
                cc = to_do_map[future]  # <16>
                print("*** Error for {}: {}".format(cc, error_msg))
            else:
                msg = "{} result: {!r}"
                print(msg.format(future, res))
                results.append(res)

    return len(results)

這里我們用到了一個對 futures.as_completed 函數(shù)特別有用的慣用法：構(gòu)建一個字典，把各個future映射到其他數(shù)據(jù)（future運(yùn)行結(jié)束后可能用的）上。這樣，雖然 future生成的順序雖然已經(jīng)亂了，依然便于使用結(jié)果做后續(xù)處理。

一篇寫完了沒有總結(jié)總感覺少點什么，所以。

Python 自 0.9.8 版就支持線程了，concurrent.futures 只不過是使用線程的最新方式。

futures.ThreadPoolExecutor 類封裝了 threading 模塊的組件，使使用線程變得更加方便。

順便再推薦一下 《流暢的python》，絕對值得一下。

下一篇筆記應(yīng)該是使用 asyncio 處理并發(fā)。

最后，感謝女朋友支持。