不論是哪一種語言，并發(fā)編程都是一項(xiàng)非常重要的技巧。比如我們上一章用的爬蟲，就被廣泛用在工業(yè)的各個領(lǐng)域。我們每天在各個網(wǎng)站、App上獲取的新聞信息，很大一部分都是通過并發(fā)編程版本的爬蟲獲得的。

正確并合理的使用并發(fā)編程，無疑會給我們的程序帶來極大性能上的提升。今天我們就一起學(xué)習(xí)Python中的并發(fā)編程——Futures。

區(qū)分并發(fā)和并行

我們在學(xué)習(xí)并發(fā)編程時，常常會聽到兩個詞：并發(fā)(Concurrency)和并行(Parallelism)這兩個術(shù)語。這兩者經(jīng)常一起使用，導(dǎo)致很多人以為他們是一個意思，其實(shí)是不對的。

首先要辨別一個誤區(qū)，在Python中，并發(fā)并不是只同一時刻上右多個操作(thread或者task)同時進(jìn)行。相反，在某個特定的時刻上它只允許有一個操作的發(fā)生，只不過線程或任務(wù)之間會相互切換直到完成，就像下面的圖里表達(dá)的

在上圖中出現(xiàn)了task和thread兩種切換順序的不同方式。分別對應(yīng)了Python中并發(fā)兩種形式——threading和asyncio。

對于線程，操作系統(tǒng)知道每個線程的所有信息，因此他會做主在適當(dāng)?shù)臅r候做線程切換，這樣的好處就是代碼容易編寫，因?yàn)槌绦騿T不需要做任何切換操作的處理；但是切換線程的操作，有可能出現(xiàn)在一個語句的執(zhí)行過程中( 比如X+=1)，這樣比較容易出現(xiàn)race condiiton的情況。

而對于asyncio，主程序想要切換任務(wù)的時候必須得到此任務(wù)可以被切換的通知，這樣一來就可以避免出現(xiàn)上面的race condition的情況。

至于所謂的并行，只在同一時刻、同時發(fā)生。Python中的multi-Processing便是這個意思對應(yīng)多進(jìn)程，我們可以這么簡單的理解，如果我們的電腦是8核的CPU，那么在運(yùn)行程序時，我們可以強(qiáng)制Python開啟8個進(jìn)程，同時執(zhí)行，用以加快程序的運(yùn)行速度。大概是下面這個圖的思路

對比看來，并發(fā)通常用于I/O操作頻繁的場景。比方我們要從網(wǎng)站上下載多個文件，由于I/O操作的時間要比CPU操作的時長多的多，這時并發(fā)就比較適合。而在CPU使用比較heavy的場景中，為了加快運(yùn)行速度，我們會多用幾臺機(jī)器，讓多個處理器來運(yùn)算。

還記得以前寫了個博客總結(jié)過：在Python中的多線程是依靠CPU切換上下文實(shí)現(xiàn)的一種“偽多線程”，在進(jìn)行大量線程切換過程中會占用比較多的CPU資源，而在進(jìn)行IO操作時候（不論是在網(wǎng)絡(luò)上進(jìn)行數(shù)據(jù)交互還是從內(nèi)存、硬盤上讀寫數(shù)據(jù)）是不需要CPU進(jìn)行計(jì)算的。所以多線程只適用于IO操作密集的環(huán)境，不適用于計(jì)算密集型操作。

并發(fā)編程之Futures

單線程于多線程性能比較

我們下面通過一個實(shí)例，從代碼的角度來理解并發(fā)編程中的Futures，并進(jìn)一步比較其于單線程的性能區(qū)別

假設(shè)我們有個任務(wù)，從網(wǎng)站上下載一些內(nèi)容然后打印出來，如果用單線程的方式是這樣實(shí)現(xiàn)的

import requests
import time
def download_one(url):
    resp = requests.get(url)
    print('Read {} from {}'.format(len(resp.content),url))
def download_all(urls):
    for url in urls:
        download_one(url)
def main():
    sites = [
        'https://en.wikipedia.org/wiki/Portal:Arts',
        'https://en.wikipedia.org/wiki/Portal:History',
        'https://en.wikipedia.org/wiki/Portal:Society', 
        'https://en.wikipedia.org/wiki/Portal:Biography',
        'https://en.wikipedia.org/wiki/Portal:Mathematics',
        'https://en.wikipedia.org/wiki/Portal:Technology',
        'https://en.wikipedia.org/wiki/Portal:Geography',
        'https://en.wikipedia.org/wiki/Portal:Science',
        'https://en.wikipedia.org/wiki/Computer_science',
        'https://en.wikipedia.org/wiki/Python_(programming_language)',
        'https://en.wikipedia.org/wiki/Java_(programming_language)',
        'https://en.wikipedia.org/wiki/PHP',
        'https://en.wikipedia.org/wiki/Node.js',
        'https://en.wikipedia.org/wiki/The_C_Programming_Language',
        'https://en.wikipedia.org/wiki/Go_(programming_language)' 
    ]
    start_time = time.perf_counter()
    download_all(sites)
    end_time = time.perf_counter()
    print('Download {} sites in {} seconds'.format(len(sites),end_time-start_time))
if __name__ == '__main__':
    main()

這是種最簡單暴力最直接的方式：

先遍歷存儲網(wǎng)站的列表

對當(dāng)前的網(wǎng)站進(jìn)行下載操作

當(dāng)前操作完成后，再對下一個網(wǎng)站進(jìn)行同樣的操作，一直到結(jié)束。

可以試出來總耗時大概是2s多，單線程的方式簡單明了，但是最大的問題是效率低下，程序最大的時間都消耗在I/O等待上（這還是用的print，如果是寫在硬盤上的話時間會更多）。如果在實(shí)際生產(chǎn)環(huán)境中，我們需要訪問的網(wǎng)站至少是以萬為單位的，所以這個方案根本行不通。

接著我們看看多線程版本的代碼

import concurrent.futures
import requests
import threading
import time
def download_one(url):
    resp = requests.get(url).content
    print('Read {} from {}'.format(len(resp),url))
def download_all(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_one,sites)
def main():
    sites = [
    'https://en.wikipedia.org/wiki/Portal:Arts',
    'https://en.wikipedia.org/wiki/Portal:History',
    'https://en.wikipedia.org/wiki/Portal:Society', 
    'https://en.wikipedia.org/wiki/Portal:Biography',
    'https://en.wikipedia.org/wiki/Portal:Mathematics',
    'https://en.wikipedia.org/wiki/Portal:Technology',
    'https://en.wikipedia.org/wiki/Portal:Geography',
    'https://en.wikipedia.org/wiki/Portal:Science',
    'https://en.wikipedia.org/wiki/Computer_science',
    'https://en.wikipedia.org/wiki/Python_(programming_language)',
    'https://en.wikipedia.org/wiki/Java_(programming_language)',
    'https://en.wikipedia.org/wiki/PHP',
    'https://en.wikipedia.org/wiki/Node.js',
    'https://en.wikipedia.org/wiki/The_C_Programming_Language',
    'https://en.wikipedia.org/wiki/Go_(programming_language)' 
    ]
    start_time = time.perf_counter()
    download_all(sites)
    # for i in sites:
    end_time = time.perf_counter()
    # print('Down {} sites in {} seconds'.format(len(sites),end_time-start_time))
if __name__ == '__main__':
    main()

這段代碼的運(yùn)行時長大概是0.2s，效率一下提升了10倍多，可以注意到這個版本和單線程的區(qū)別主要在下面：

def download_all(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_one,sites)

在上面的代碼中我們創(chuàng)建了一個線程池，有5個線程可以分配使用。executer.map()與以前將的Python內(nèi)置的map()函數(shù)，表示對sites中的每一個元素并發(fā)的調(diào)用函數(shù)download_one()函數(shù)。

順便提一下，在download_one()函數(shù)中，我們使用的requests.get()方法是線程安全的(thread-safe)，因此在多線程的環(huán)境下，它也可以安全使用，并不會出現(xiàn)race condition（條件競爭）的情況。

另外，雖然線程的數(shù)量可以自己定義，但是線程數(shù)并不是越多越好，以為線程的創(chuàng)建、維護(hù)和刪除也需要一定的開銷。所以如果設(shè)置的很大，反而會導(dǎo)致速度變慢，我們往往要根據(jù)實(shí)際的需求做一些測試，來尋找最優(yōu)的線程數(shù)量。

當(dāng)然，我們也可以用并行的方式去提高運(yùn)行效率，只需要在download_all()函數(shù)中做出下面的變化即可

def download_all(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        to_do = []
        for site in sites:
            future = executor.submit(download_one,site)
            to_do.append(site)

        for future in concurrent.futures.as_completed(to_do):
            future.result()

在需要改的這部分代碼中，函數(shù)ProcessPoolExecutor()表示創(chuàng)建進(jìn)程池，使用多個進(jìn)程并行的執(zhí)行程序。不過，這里 通常省略參數(shù)workers，因?yàn)橄到y(tǒng)會自動返回CPU的數(shù)量作為可以調(diào)用的進(jìn)程數(shù)。

就像上面說的，并行方式一般用在CPU密集型的場景中，因?yàn)閷τ贗/O密集型操作多數(shù)時間會用于等待，相比于多線程，使用多進(jìn)程并不會提升效率，反而很多時候，因?yàn)镃PU數(shù)量的限制，會導(dǎo)致執(zhí)行效率不如多線程版本。

到底什么是Futures？

Python中的Futures，位于concurrent.futures和asyncio中，他們都表示帶有延遲的操作，F(xiàn)utures會將處于等待狀態(tài)的操作包裹起來放到隊(duì)列中，這些操作的狀態(tài)可以隨時查詢。而他們的結(jié)果或是異常，也能在操作后被獲取。

通常，作為用戶，我們不用考慮如何去創(chuàng)建Futures，這些Futures底層會幫我們處理好，我們要做的就是去schedule這些Futures的執(zhí)行。比方說，F(xiàn)utures中的Executor類，當(dāng)我們中的方法done()，表示相對應(yīng)的操作是否完成——用True表示已完成，ongFalse表示未完成。不過，要注意的是done()是non-blocking的，會立刻返回結(jié)果，相對應(yīng)的add_done_callback(fn)，則表示Futures完成后，相對應(yīng)的參數(shù)fn，會被通知并執(zhí)行調(diào)用。

Futures里還有一個非常重要的函數(shù)result()，用來表示future完成后，返回器對應(yīng)的結(jié)果或異常。而as_completed(fs)，則是針對給定的future迭代器fs，在其完成后，返回完成后的迭代器。

所以也可以把上面的例子寫成下面的形式：

def download_all(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        to_do = []
        for site in sites:
            future = executor.submit(download_one,site)
            to_do.append(site)
        for future in concurrent.futures.as_completed(to_do):
            future.result()

這里，我們首先用executor.submit()，將下載每個網(wǎng)站的內(nèi)容都放進(jìn)future隊(duì)列to_do里等待執(zhí)行。然后是as_completed()函數(shù)，在future完成后輸出結(jié)果

不過這里有個事情要注意一下：future列表中每個future完成的順序和他在列表中的順序不一定一致，至于哪個先完成，取決于系統(tǒng)的調(diào)度和每個future的執(zhí)行時間。

為什么多線程每次只有一個線程執(zhí)行？

前面我們講過，在一個時刻下，Python主程序只允許有一個線程執(zhí)行，所以Python的并發(fā)，是通過多線程的切換完成的，這是為什么呢？

這就又和以前講的知識串聯(lián)到一起了——GIL（全局解釋器鎖），這里在復(fù)習(xí)下：

事實(shí)上，Python的解釋器并不是線程安全的，為了解決由此帶來的race condition等問題，Python就引入了GIL，也就是在同一個時刻，只允許一個線程執(zhí)行。當(dāng)然，在進(jìn)行I/O操作是，如果一個線程被block了，GIL就會被釋放，從而讓另一個線程能夠繼續(xù)執(zhí)行。

總結(jié)

這節(jié)課里我們先學(xué)習(xí)了Python中并發(fā)和并行的概念

并發(fā)——通過線程(thread)和任務(wù)(task)之間相互切換的方式實(shí)現(xiàn)，但是同一時刻，只允許有一個線程或任務(wù)執(zhí)行

并行——多個進(jìn)程同時進(jìn)行。

并發(fā)通常用于I/O頻繁操作的場景，而并行則適用于CPU heavy的場景

隨后我們通過一個下載網(wǎng)站內(nèi)容的例子，比較了單線程和運(yùn)用FUtures的多線程版本的性能差異，顯而易見，合理的運(yùn)用多線程，能夠極大的提高程序運(yùn)行效率。

我們還大致了解了Futures的方式，介紹了一些常用的函數(shù)，并輔以實(shí)例加以理解。

要注意，Python中之所以同一時刻只允許一個線程運(yùn)行，其實(shí)是由于GIL的存在。但是對于I/O操作而言，當(dāng)其被block的時候，GIL會被釋放，使其他線程繼續(xù)執(zhí)行。

以上就是Python并發(fā)編程之未來模塊Futures的詳細(xì)內(nèi)容，更多關(guān)于Python并發(fā)未來模塊Futures的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論