Python并發(fā)和并行方案

在Python世界有3種并發(fā)和并行方案，如下:

多線程(threading)

多進(jìn)程(multiprocessing)

異步IO(asyncio)

注: 并發(fā)和并行的區(qū)別先不提，最后會(huì)借著例子更好的解釋?zhuān)硗馍院笠矔?huì)提到 concurrent.futures，不過(guò)它不是一種獨(dú)立的方案，所以在這里沒(méi)有列出來(lái)。

這些方案是為了解決不同特點(diǎn)的性能瓶頸。性能問(wèn)題主要有2種:

CPU密集型(CPU-bound)。這也就是指計(jì)算密集型任務(wù)，它的特點(diǎn)是需要要進(jìn)行大量的計(jì)算。例如Python內(nèi)置對(duì)象的各種方法的執(zhí)行，科學(xué)計(jì)算，視頻轉(zhuǎn)碼等等。

I/O密集型(I/O-bound)。凡是涉及到網(wǎng)絡(luò)、內(nèi)存訪問(wèn)、磁盤(pán)I/O等的任務(wù)都是IO密集型任務(wù)，這類(lèi)任務(wù)的特點(diǎn)是CPU消耗很少，任務(wù)的大部分時(shí)間都在等待I/O操作完成。例如數(shù)據(jù)庫(kù)連接、Web服務(wù)、文件讀寫(xiě)等等。

如果你不知道一個(gè)任務(wù)哪種類(lèi)型，我的經(jīng)驗(yàn)是你問(wèn)問(wèn)自己，如果給你一個(gè)更好更快的CPU它可以更快，那么這就是一個(gè)CPU密集的任務(wù)，否則就是I/O密集的任務(wù)。

這三個(gè)方案中對(duì)于CPU密集型的任務(wù)，優(yōu)化方案只有一種，就是使用多進(jìn)程充分利用多核CPU一起完成任務(wù)，達(dá)到提速的目的。而對(duì)于I/O密集型的任務(wù)，則這三種方案都可以。

接著借著一個(gè)抓取網(wǎng)頁(yè)并寫(xiě)入本地(典型的I/O密集型任務(wù))小例子來(lái)挨個(gè)拆解對(duì)比一下這些方案。先看例子:

import requests

url?
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36' # noqa
}


def fetch(session, page):
with (session.get(f'{url}{page*25}', headers=headers) as r,
open(f'top250-{page}.html', 'w') as f):
f.write(r.text)


def main():
with requests.Session() as session:
for p in range(25):
fetch(session, p)


if __name__ == '__main__':
main()
在這個(gè)例子中會(huì)抓取豆瓣電影Top250的25個(gè)頁(yè)面(每頁(yè)顯示10個(gè)電影)，使用requests庫(kù)，不同頁(yè)面按順序請(qǐng)求，一共花了3.9秒:

? time python io_non_concurrent.py
python io_non_concurrent.py 0.23s user 0.05s system 7% cpu 3.911 total
這個(gè)速度雖然看起來(lái)還是很好的，一方面是豆瓣做了很好的優(yōu)化，一方面我家的帶寬網(wǎng)速也比較好。接著用上面三種方案優(yōu)化看看效果。

多進(jìn)程版本
Python解釋器使用單進(jìn)程，如果服務(wù)器或者你的電腦是多核的，這么用其實(shí)是很浪費(fèi)的，所以可以通過(guò)多進(jìn)程提速:

from multiprocessing import Pool

def main():
with (Pool() as pool,
requests.Session() as session):
pool.starmap(fetch, [(session, p) for p in range(25)])
注: 這里省略到了那些上面已經(jīng)出現(xiàn)的了代碼，只展示改變了的那部分。

使用多進(jìn)程池，但沒(méi)指定進(jìn)程數(shù)量，所以會(huì)按著Macbook的核數(shù)啟動(dòng)10個(gè)進(jìn)程一起工作，耗時(shí)如下:

? time python use_multiprocessing.py
python use_multiprocessing.py 2.15s user 0.30s system 232% cpu 1.023 total
多進(jìn)程理論上可以有十倍效率的提升，因?yàn)?0個(gè)進(jìn)程在一起執(zhí)行任務(wù)。當(dāng)然由于任務(wù)數(shù)量是25，不是整數(shù)倍，是無(wú)法達(dá)到10倍的降低耗時(shí)，而且由于抓取太快了，沒(méi)有充分顯示多進(jìn)程方案下的效率提升，所以用時(shí)1秒，也就是大約4倍的效率提升。

多進(jìn)程方案下沒(méi)有明顯的缺點(diǎn)，只要機(jī)器夠強(qiáng)悍，就可以更快。

多線程版本
Python解釋器不是線程安全的，為此Python設(shè)計(jì)了GIL: 獲得GIL鎖才可以訪問(wèn)線程中的Python對(duì)象。所以在任何一個(gè)時(shí)間，只有一個(gè)線程可以執(zhí)行代碼，這樣就不會(huì)引發(fā)競(jìng)態(tài)條件(Race Condition) ，雖然GIL的問(wèn)題很多，但是GIL卻是還有它存在的優(yōu)點(diǎn)，例如簡(jiǎn)化了內(nèi)存管理等等，這些不是本文重點(diǎn)所以就不展開(kāi)了，有興趣的可以專(zhuān)門(mén)去了解。

那么有同學(xué)會(huì)問(wèn)，既然同一時(shí)間永遠(yuǎn)只有一個(gè)線程在工作，那么多線程可以提高并發(fā)效率的原因是什么呢？

解釋這個(gè)問(wèn)題還是要提GIL。延伸閱讀鏈接1《Understanding the Python GIL》中做了很好的解釋?zhuān)ㄟ@里要注意，我們提的方案是Python 3.2新的GIL，而不是Python2的舊版GIL，現(xiàn)在網(wǎng)上有很多針對(duì)舊的GIL的描述，其實(shí)是過(guò)時(shí)的，這部分也可以看看延伸閱讀鏈接2的文章幫助理解它們的區(qū)別），我截其中幾張PPT來(lái)說(shuō)明:


在上圖里，本來(lái)只有1個(gè)線程，所以不需要釋放或者獲得GIL，但是接著出現(xiàn)了第二個(gè)線程，這樣就是多個(gè)線程，一開(kāi)始線程2是掛起狀態(tài)，因?yàn)樗鼪](méi)有GIL。

線程1在一個(gè) cv_wait周期內(nèi)會(huì)自愿的放棄GIL，例如出現(xiàn)了I/O阻塞，或者超時(shí)了(線程不能一直拿著不放，即便在一個(gè)周期內(nèi)沒(méi)有出現(xiàn)I/O阻塞也要強(qiáng)制釋放執(zhí)行權(quán)，這個(gè)默認(rèn)時(shí)間是5毫秒，可以通過(guò) sys.setswitchinterval設(shè)置，當(dāng)然設(shè)置前你得知道你在做什么)都會(huì)觸發(fā)這個(gè)釋放GIL的操作。

這里演示了常規(guī)的例子(非超時(shí)被迫釋放)，在 cv_wait階段，線程1由于遇到了I/O阻塞，會(huì)發(fā)送信號(hào)給線程2，此時(shí)線程1讓出GIL并掛起，而線程2獲得GIL，如此循環(huán)，之后線程2會(huì)釋放GIL給線程1。這個(gè)PPT在業(yè)界非常知名，建議大家多看看。之后的PPT還列舉了超時(shí)的處理，由于和我們這篇文章關(guān)系稍遠(yuǎn)也不展開(kāi)了，有興趣的接著看。btw，我第一次看這個(gè)PPT覺(jué)得這個(gè)超時(shí)時(shí)間好可怕，也就是說(shuō)1秒鐘要最少切換200次，這也太浪費(fèi)了，所以你可以嘗試在代碼中調(diào)大這個(gè)超時(shí)時(shí)間喲。

通過(guò)上面的內(nèi)容，多線程通過(guò)GIL的控制，每個(gè)線程都得到了更好的執(zhí)行時(shí)機(jī)，所以不會(huì)出現(xiàn)被某個(gè)線程任務(wù)一直阻塞，因?yàn)槿绻€程遇到阻塞會(huì)自愿讓出GIL讓自己掛起，把機(jī)會(huì)讓給其他線程，這樣就提高了執(zhí)行任務(wù)總體的效率。多線程模式下最完美的場(chǎng)景就是任何時(shí)間點(diǎn)對(duì)應(yīng)的線程都在做事，而不是有的線程其實(shí)等著被執(zhí)行，但是實(shí)際上卻被阻塞著。

我們看一下多線程的方案:

from multiprocessing.pool import ThreadPool

def main():
with (ThreadPool(processes=5) as pool,
requests.Session() as session):
pool.starmap(fetch, [(session, p) for p in range(25)])
這里說(shuō)明2點(diǎn):

多進(jìn)程和多線程例子中我都使用了【池】，這是一個(gè)好的習(xí)慣，因?yàn)榫€(進(jìn))程過(guò)多會(huì)帶來(lái)額外的開(kāi)銷(xiāo)，其中包括創(chuàng)建銷(xiāo)毀的開(kāi)銷(xiāo)、調(diào)度開(kāi)銷(xiāo)等等，同時(shí)也降低了計(jì)算機(jī)的整體性能。使用線(進(jìn))程池維護(hù)多個(gè)線(進(jìn))程，等待監(jiān)督管理者分配可并發(fā)執(zhí)行的任務(wù)。這樣一方面避免了處理任務(wù)時(shí)創(chuàng)建銷(xiāo)毀線程開(kāi)銷(xiāo)的代價(jià)，另一方面避免了線程數(shù)量膨脹導(dǎo)致的過(guò)分調(diào)度問(wèn)題，保證了對(duì)內(nèi)核的充分利用。另外用標(biāo)準(zhǔn)庫(kù)里的進(jìn)程池和線程池的實(shí)現(xiàn)寫(xiě)額外代碼極少，而且代碼結(jié)構(gòu)還很像，特別適合寫(xiě)對(duì)比的例子。

processes如果不指定也是和CPU核數(shù)一致的10，但是并不是線程越多越好，因?yàn)榫€程多了，反而出現(xiàn)本來(lái)正常有效的執(zhí)行卻被GIL強(qiáng)制釋放，這就造成多余上下文切換反而是一個(gè)負(fù)擔(dān)了。

在這個(gè)例子中，線程數(shù)為5，這個(gè)其實(shí)一方面是經(jīng)驗(yàn)，一方面是多次調(diào)試值的結(jié)果，所以這也暴露了多線程編程中如果稍有不慎會(huì)讓優(yōu)化變差，也會(huì)存在沒(méi)有找到最優(yōu)值得問(wèn)題，因?yàn)镚IL控制線程是一個(gè)黑盒操作，開(kāi)發(fā)者無(wú)法直接控制，這哪怕對(duì)一些相對(duì)有經(jīng)驗(yàn)的Python開(kāi)發(fā)也非常不友好。

我們看一下時(shí)間:

? time python use_threading.py
python use_threading.py 0.62s user 0.24s system 74% cpu 1.157 total
可以看到，多線程方案下比原始方案速度快了一倍以上，但是比多進(jìn)程方案差一點(diǎn)(事實(shí)上我認(rèn)為在真實(shí)的例子中會(huì)差很多)。這是因?yàn)樵诙噙M(jìn)程方案下多核CPU都在獨(dú)立工作，而多線程方案一方面由于效率問(wèn)題下不能使用那么多數(shù)量的線程，而且由于GIL的限制，在不需要被釋放GIL的時(shí)候依然被強(qiáng)制釋放，就這么不斷的切換的過(guò)程中反而降低了效率，讓效果大打折扣。

concurrent.futures版本
這里也順便提一下 concurrent.futures的方案。其實(shí)它不是一個(gè)全新的方案，這是在其他語(yǔ)言(例如Java)里早就出現(xiàn)的一種框架，可以通過(guò)它控制線(進(jìn))程的啟動(dòng)、執(zhí)行和關(guān)閉。我把它理解為抽象了多進(jìn)程池和多線程池的代碼，讓開(kāi)發(fā)者不需要關(guān)注多線程和多進(jìn)程模塊的具體細(xì)節(jié)和用法。其實(shí)理解起來(lái)也不難，你可以這么拆解:

其實(shí)理解起來(lái)也不難，例如ThreadPoolExecutor可以這么拆解: ThreadPoolExecutor=Thread+Pool+Executor，其實(shí)就是線程+池+執(zhí)行器。就是預(yù)先創(chuàng)建一個(gè)線程池用來(lái)被重復(fù)使用，Executor將任務(wù)提交和任務(wù)執(zhí)行進(jìn)行解耦，它完成線程的調(diào)配(如何以及何時(shí))和任務(wù)的執(zhí)行部分。

如果你想了解它的細(xì)節(jié)，我推薦直接看它的源碼文件頭部的注釋?zhuān)锩鎸?duì)于數(shù)據(jù)流有非常詳細(xì)的說(shuō)明，可以說(shuō)比任何技術(shù)文章寫(xiě)的都要深入準(zhǔn)確了。

這里只演示一下ThreadPoolExecutor的用法:

from functools import partial
from concurrent.futures import ThreadPoolExecutor

def main():
with (ThreadPoolExecutor(max_workers=5) as pool,
requests.Session() as session):
list(pool.map(partial(fetch, session), range(25)))
是不是很熟悉的配方？接口和上面用的進(jìn)程池線程池都很像，但是要注意 max_workers如果不指定的話數(shù)量是CPU個(gè)數(shù)+4，最大為32。它和多線程的用法問(wèn)題一樣，這個(gè) max_workers需要調(diào)優(yōu)(這里為了對(duì)比，所以用了相同的數(shù)值）。

? time python use_executor.py
python use_executor.py 0.63s user 0.32s system 82% cpu 1.153 total
雖然 concurrent.futures是現(xiàn)在更主流的方案，但是在我使用的體驗(yàn)里，它的效率要略低于直接使用進(jìn)程池或者線程池的代碼，因?yàn)樗叨瘸橄?，卻把事情搞得復(fù)雜了，例如用到了對(duì)應(yīng)的queue(queue模塊)和信號(hào)量(Semaphore)，這些反而限制了性能的提升。所以我的建議是，Python初學(xué)者可以用它，但高級(jí)開(kāi)發(fā)者應(yīng)該自己控制并發(fā)實(shí)現(xiàn)。

asyncio版本
前面的多線程相關(guān)的方案中，需要開(kāi)發(fā)者根據(jù)經(jīng)驗(yàn)或者去實(shí)驗(yàn)，找到一個(gè)(或者多個(gè))最優(yōu)的線程數(shù)量，不同的場(chǎng)景這個(gè)值區(qū)別是很大的，這對(duì)于初學(xué)者很不友好，非常容易陷入【在用多線程，但是用錯(cuò)了或者用的不夠好】這么一種境地。

后來(lái)Python引入了新的并發(fā)模型: aysncio，本小節(jié)給大家解釋下最新的asyncio方案為什么是一個(gè)更優(yōu)的選擇。首先還是看《Understanding the Python GIL》里面的一頁(yè)P(yáng)PT:

我們回憶一下，它提到當(dāng)只有單個(gè)線程時(shí)，實(shí)際上不會(huì)觸發(fā)GIL，這個(gè)獨(dú)立的線程可以一直執(zhí)行下去。這也是asyncio找到的切入點(diǎn): 因?yàn)槭菃芜M(jìn)程單線程的，所以理論上不受GIL的限制。在事件驅(qū)動(dòng)的機(jī)制下，可以更好的利用單線程的性能，尤其是通過(guò)await關(guān)鍵詞可以讓開(kāi)發(fā)者自己決定調(diào)度方案，而不是多線程那種由GIL來(lái)控制。

那設(shè)想一下，在最美好的情況下，所有await的地方都是可能的I/O阻塞的。那么在執(zhí)行時(shí)，遇到I/O阻塞就可以切換協(xié)程，執(zhí)行其他可以繼續(xù)執(zhí)行的任務(wù)，所以，這個(gè)線程一直都在工作而不會(huì)阻塞，可以說(shuō)利用率達(dá)到100%！這是多線程方案下永遠(yuǎn)不可及的。

講到這個(gè)，我們?cè)倩厝ブ匦抡砗屠斫庖槐椋瘸龌纠碚撻_(kāi)始。

協(xié)程

協(xié)程是一種特殊函數(shù)，這個(gè)函數(shù)在本來(lái)的def關(guān)鍵字前面加了async關(guān)鍵字，本質(zhì)上它是生成器函數(shù)，可以生成值或者接收外面發(fā)送(通過(guò)send方法)來(lái)的值，但是它最重要的特點(diǎn)是它可以在需要時(shí)保存上下文(或者說(shuō)狀態(tài))，掛起自己并將控制權(quán)交給調(diào)用者，由于它保存了掛起時(shí)的上下文，在未來(lái)可以接著被執(zhí)行。

其實(shí)在調(diào)用協(xié)程是，它并不會(huì)立刻執(zhí)行:

In : async def a():

...:? ? ?print('Called')

...:

In : a()? # 并未執(zhí)行，只是返回了協(xié)程對(duì)象

Out:

In : await a()? # 使用await才會(huì)真的執(zhí)行

Called

異步和并發(fā)

異步(asynchronous)、非阻塞(non-blocking)、并發(fā)(concurrent)是很容易讓人產(chǎn)生迷惑的詞。結(jié)合asyncio場(chǎng)景，我的理解是:

協(xié)程是異步執(zhí)行的，在asyncio中，協(xié)程可以在等待執(zhí)行結(jié)果時(shí)把自己【暫?！浚员阕屍渌麉f(xié)程同時(shí)運(yùn)行。

異步讓執(zhí)行不需要等待阻塞的邏輯完成就可以先讓其他代碼同時(shí)運(yùn)行，所以這樣就不會(huì)【阻塞】其他代碼，那么這就是【非阻塞】的代碼

使用異步代碼編寫(xiě)的程序執(zhí)行時(shí)，看起來(lái)其中的任務(wù)都在同時(shí)執(zhí)行和完成(因?yàn)闀?huì)在等待中切換)，所以看起來(lái)是【并發(fā)】的

事件循環(huán)(EventLoop)

Event Loop這個(gè)概念其實(shí)我理解了很多年，從Twisted時(shí)代開(kāi)始。我一直覺(jué)得它非常神秘復(fù)雜，現(xiàn)在看來(lái)其實(shí)想多了。對(duì)于初學(xué)者，不如換個(gè)思路，它的重點(diǎn)就是事件+循環(huán): Loop是一個(gè)環(huán)，每個(gè)任務(wù)作為一個(gè)事件放到這個(gè)環(huán)上，事件會(huì)不斷地循環(huán)，在符合條件的情況下觸發(fā)執(zhí)行事件。它的特點(diǎn)如下:

一個(gè)事件循環(huán)運(yùn)行在一個(gè)線程中

Awaitables對(duì)象(協(xié)程、Task、Future下面都會(huì)提到)都可以注冊(cè)到事件循環(huán)上

如果協(xié)程中調(diào)用了另外一個(gè)協(xié)程(通過(guò)await),這個(gè)協(xié)程會(huì)掛起，發(fā)生上下文切換轉(zhuǎn)而去執(zhí)行另外這個(gè)協(xié)程，如此循環(huán)

如果協(xié)程執(zhí)行時(shí)遇到I/O阻塞，這個(gè)協(xié)程會(huì)帶著上下文掛起，然后把控制權(quán)交還給EventLoop

既然是loop。注冊(cè)的全部事件執(zhí)行完畢后，循環(huán)會(huì)重新開(kāi)始

Future/Task

asyncio.Future我覺(jué)得像Javascript里面的 Promise, 它是一個(gè)占位對(duì)象，代表一件還沒(méi)有做完的事情，在未來(lái)才會(huì)實(shí)現(xiàn)或者完成(當(dāng)然還可能由于內(nèi)部出錯(cuò)而拋出異常)。它和上面提的 concurrent.futures方案中實(shí)現(xiàn)的 concurrent.futures.Futures很像，但是針對(duì)asyncio的事件循環(huán)做了很多定制。asyncio.Future它僅僅是一個(gè)數(shù)據(jù)的容器。

asyncio.Task是 asyncio.Future的子類(lèi)，它用于在事件循環(huán)中運(yùn)行協(xié)程。

在官方文檔中提到了一個(gè)非常直觀的例子，我這里改寫(xiě)它在IPython里面執(zhí)行并說(shuō)明:

In : async def set_after(fut):? # 創(chuàng)建一個(gè)協(xié)程，他會(huì)異步的sleep3秒，然后給future對(duì)象設(shè)置結(jié)果

...:? ? ?await asyncio.sleep(3)

...:? ? ?fut.set_result('Done')

...:

In : loop = asyncio.get_event_loop()? # 獲取當(dāng)前的事件循環(huán)

In : fut = loop.create_future()? # 在事件循環(huán)中創(chuàng)建一個(gè)Future

In : fut? # 此時(shí)它還是默認(rèn)的pending狀態(tài)，因?yàn)闆](méi)有調(diào)用它

Out:

In : task = loop.create_task(set_after(fut))? # 在事件循環(huán)中創(chuàng)建(或者說(shuō)注冊(cè))了一個(gè)任務(wù)

In : task? # 馬上輸入它，此時(shí)剛創(chuàng)建任務(wù)，還在執(zhí)行中

Out: :2> wait_for=()]>>

In : fut? # 馬上輸入它，此時(shí)剛創(chuàng)建任務(wù)，還沒(méi)有執(zhí)行完所以future沒(méi)有變化

Out:

In : task? # 過(guò)了三秒，任務(wù)執(zhí)行完成了

Out: :1> result=None>

In : fut? # Future也已經(jīng)設(shè)置了結(jié)果，所以狀態(tài)是finished

Out:

可以感受到:

Future對(duì)象不是任務(wù)，就是存放狀態(tài)的一個(gè)容器

create_task會(huì)讓事件循環(huán)調(diào)度協(xié)程的執(zhí)行

創(chuàng)建任務(wù)可以用 ensure_future和 create_task， ensure_future是一個(gè)更高級(jí)封裝的函數(shù)，但是Python3.7以上版本應(yīng)該使用 create_task

接著是了解await的作用。如果協(xié)程中await一個(gè)Future對(duì)象，Task會(huì)暫停協(xié)程的執(zhí)行并等待Future的完成。而當(dāng)Future完成后，包裝協(xié)程的執(zhí)行將繼續(xù):

In : async def a():

...:? ? ?print('IN a')

...:? ? ?await b()

...:? ? ?await c()

...:? ? ?print('OUT a')

...:

In : async def b():

...:? ? ?print('IN b')

...:? ? ?await d()

...:? ? ?print('OUT b')

...:

...:

In : async def c():

...:? ? ?print('IN c')

...:? ? ?await asyncio.sleep(1)

...:? ? ?print('OUT c')

...:

...:

In : async def d():

...:? ? ?print('IN d')

...:? ? ?await asyncio.sleep(1)

...:? ? ?print('OUT d')

...:

In : asyncio.run(a())

IN a

IN b

IN d

OUT d

OUT b

IN c

OUT c

OUT a

這個(gè)例子中，a的入口函數(shù)，其中調(diào)用b和c，b又會(huì)調(diào)用d。await會(huì)讓對(duì)應(yīng)的協(xié)程獲取執(zhí)行權(quán)限，協(xié)程內(nèi)await的其他協(xié)程都執(zhí)行完畢才會(huì)釋放權(quán)限，所以注意這個(gè)更像DFS(深度優(yōu)先搜索)，所以執(zhí)行順序是a->b->d->c。

所以這里就得出結(jié)論:

事件循環(huán)負(fù)責(zé)協(xié)程的協(xié)作調(diào)度：事件循環(huán)一次運(yùn)行一個(gè)任務(wù)。當(dāng)一個(gè)任務(wù)等待一個(gè)Awaitables對(duì)象完成時(shí)，事件循環(huán)會(huì)運(yùn)行其他任務(wù)、回調(diào)或執(zhí)行 IO 操作。

asyncio方案

在asyncio方案里，凡是涉及I/O阻塞操作的庫(kù)都要使用aio生態(tài)中的庫(kù)，所以已經(jīng)不能再使用requests庫(kù)，而是需要使用aiohttp，另外文件操作需要使用aiofiles。最終代碼如下(這個(gè)2個(gè)包需要下載再使用):

import aiofiles

import asyncio

import aiohttp

async def fetch(session, page):

? ? r = await session.get(f'{url}{page*25}', headers=headers)

? ? async with aiofiles.open(f'top250-{page}.html', mode='w') as f:

? ? ? ? await f.write(await r.text())

async def main():

? ? loop = asyncio.get_event_loop()

? ? async with aiohttp.ClientSession(loop=loop) as session:

? ? ? ? tasks = [asyncio.ensure_future(fetch(session, p)) for p in range(25)]

? ? ? ? await asyncio.gather(*tasks)

if __name__ == '__main__':

? ? asyncio.run(main())

看一下效率:

? time python use_asyncio.py

python use_asyncio.py? 0.20s user 0.04s system 34% cpu 0.684 total

所以asyncio的優(yōu)點(diǎn)如下:

asyncio用好了，是這些并發(fā)方案中最快的

它支持?jǐn)?shù)千級(jí)別的活動(dòng)連接，這對(duì)于websockets和MQTT之類(lèi)的場(chǎng)景下性能可以表現(xiàn)的很好，而多線程方案中在這個(gè)規(guī)模的線程數(shù)量下會(huì)出現(xiàn)嚴(yán)重的性能問(wèn)題。

多線程方案下線程切換是隱式的，我們無(wú)法確認(rèn)它何時(shí)會(huì)切換線程的執(zhí)行權(quán)，所以非常容易出現(xiàn)競(jìng)態(tài)條件(Race Condition)。而asyncio方案里協(xié)程的切換是顯式、明確的，開(kāi)發(fā)者可以明確地獲知或者指定執(zhí)行的順序

并發(fā)和并行

我之前翻到了一個(gè)對(duì)比這些方案的說(shuō)法(延伸鏈接4)，其中也提到了并發(fā)和并行，說(shuō)的特別形象，我加以說(shuō)明:

多進(jìn)程。10個(gè)廚房，10個(gè)廚子，10道菜。也是1個(gè)廚房1廚子做1道菜。

多線程。1個(gè)廚房，10個(gè)廚子，10道菜。因?yàn)閺N房比較小，只能大家一起擠在里面，事實(shí)上是輪著做，而且一個(gè)廚師在做的時(shí)候其他人只能等著輪到自己。

asyncio。1個(gè)廚房，1個(gè)廚子，10道菜。聽(tīng)起來(lái)好像這就是一個(gè)順序執(zhí)行，但事實(shí)上，當(dāng)某道菜需要燉或者其他什么耗時(shí)的烹飪方法時(shí)，可以同時(shí)做其他的菜或者做準(zhǔn)備，最美好的場(chǎng)景是這個(gè)廚師一直在忙著做。

對(duì)于并發(fā)和并行我推薦看一下延伸閱讀連接3的文章。并發(fā)(Concurrency)允許同時(shí)執(zhí)行多個(gè)任務(wù)，這些任務(wù)可能訪問(wèn)相同的共享資源，例如硬盤(pán)、網(wǎng)絡(luò)以及對(duì)應(yīng)的那個(gè)單核CPU。既然會(huì)出現(xiàn)訪問(wèn)共享資源，就可能出現(xiàn)競(jìng)態(tài)條件，所以某個(gè)時(shí)間點(diǎn)事實(shí)上只有一個(gè)任務(wù)在執(zhí)行，在本質(zhì)上目標(biāo)是當(dāng)一個(gè)任務(wù)被迫等待外部資源時(shí)，通過(guò)在它們之間切換來(lái)防止任務(wù)相互阻塞，系統(tǒng)會(huì)有機(jī)制保證這些任務(wù)都在推進(jìn)。并行(Parallelism)是指多個(gè)任務(wù)在獨(dú)立分區(qū)的資源(如多個(gè)CPU內(nèi)核)上并行運(yùn)行，這樣可以最大限度地利用硬件資源。


審核編輯：劉清

?