您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

物無定味適口者珍，Python3並發場景(CPU密集/IO密集)任務的並發方式的場景抉擇(多線程threading/多進程multiprocessing/協程asyncio)

編輯：Python

原文轉載自「劉悅的技術博客」https://v3u.cn/a_id_221

一般情況下，大家對Python原生的並發/並行工作方式：進程、線程和協程的關系與區別都能講清楚。甚至具體的對象名稱、內置方法都可以如數家珍，這顯然是極好的，但我們其實都忽略了一個問題，就是具體應用場景，三者的使用目的是一樣的，話句話說，使用結果是一樣的，都可以提高程序運行的效率，但到底那種場景用那種方式更好一點？

這就好比，目前主流的汽車發動機變速箱無外乎三種：雙離合、CVT以及傳統AT。主機廠把它們搭載到不同的發動機和車型上，它們都是變速箱，都可以將發動機產生的動力作用到車輪上，但不同使用場景下到底該選擇那種變速箱？這顯然也是一個問題。

所謂“無場景，不功能”，本次我們來討論一下，具體的並發編程場景有哪些，並且對應到具體場景，應該怎麼選擇並發手段和方式。

什麼是並發和並行？

在討論場景之前，我們需要將多任務執行的方式進行一下分類，那就是並發方式和並行方式。教科書上告訴我們：並行是指兩個或者多個事件在同一時刻發生；而並發是指兩個或多個事件在同一時間間隔內發生。在多道程序環境下，並發性是指在一段時間內宏觀上有多個程序在同時運行，但在單處理機系統中，每一時刻卻僅能有一道程序執行，故微觀上這些程序只能是分時地交替執行。

好像有那麼一點抽象，好吧，讓我們務實一點，由於GIL全局解釋器鎖的存在，在Python編程領域，我們可以簡單粗暴地將並發和並行用程序通過能否使用多核CPU來區分，能使用多核CPU就是並行，不能使用多核CPU，只能單核處理的，就是並發。就這麼簡單，是的，Python的GIL全局解釋器鎖幫我們把問題簡化了，這是Python的大幸？還是不幸？

Python中並發任務實現方式包含：多線程threading和協程asyncio，它們的共同點都是交替執行，而區別是多線程threading是搶占式的，而協程asyncio是協作式的，原理也很簡單，只有一顆CPU可以用，而一顆CPU一次只能做一件事，所以只能靠不停地切換才能完成並發任務。

Python中並行任務的實現方式是多進程multiprocessing，通過multiprocessing庫，Python可以在程序主進程中創建新的子進程。這裡的一個進程可以被認為是一個幾乎完全不同的程序，盡管從技術上講，它們通常被定義為資源集合，其中資源包括內存、文件句柄等。換一種說法是，每個子進程都擁有自己的Python解釋器，因此，Python中的並行任務可以使用一顆以上的CPU，每一顆CPU都可以跑一個進程，是真正的同時運行，而不需要切換，如此Python就可以完成並行任務。

什麼時候使用並發？IO密集型任務

現在我們搞清楚了，Python裡的並發運行方式就是多線程threading和協程asyncio，那麼什麼場景下使用它們？

一般情況下，任務場景，或者說的更准確一些，任務類型，無非兩種：CPU密集型任務和IO密集型任務。

什麼是IO密集型任務？IO就是Input-Output的縮寫，說白了就是程序的輸入和輸出，想一想確實就是這樣，您的電腦，它不就是這兩種功能嗎？用鍵盤、麥克風、攝像頭輸入數據，然後再用屏幕和音箱進行輸出操作。

但輸入和輸出操作要比電腦中的CPU運行速度慢，換句話說，CPU得等著這些比它慢的輸入和輸出操作，說白了就是CPU運算一會，就得等這些IO操作，等IO操作完了，CPU才能繼續運算一會，然後再等著IO操作，如圖所示：

由此可知，並發適合這種IO操作密集和頻繁的工作，因為就算CPU是蘋果最新ARM架構的M2芯片，也沒有用武之地。

另外，如果把IO密集型任務具象化，那就是我們經常操作的：硬盤讀寫(數據庫讀寫)、網絡請求、文件的打印等等。

並發方式的選擇：多線程threading還是協程asyncio？

既然涉及硬盤讀寫(數據庫讀寫)、網絡請求、文件打印等任務都算並發任務，那我們就真正地實踐一下，看看不同的並發方式到底能提升多少效率？

一個簡單的小需求，對本站數據進行重復抓取操作，並計算首頁數據文本的行數：

import requests

import time
def download_site(url, session):

with session.get(url) as response:

print(f"下載了{len(response.content)}行數據")
def download_all_sites(sites):

with requests.Session() as session:

for url in sites:

download_site(url, session)
if __name__ == "__main__":
sites = ["https://v3u.cn"] * 50

start_time = time.time()

download_all_sites(sites)

duration = time.time() - start_time

print(f"下載了 {len(sites)}次，執行了{duration}秒")

在不使用任何並發手段的前提下，程序返回：

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了76347行數據

下載了 50 次數據，執行了8.781155824661255秒

[Finished in 9.6s]

這裡程序的每一步都是同步操作，也就是說當第一次抓取網站首頁時，剩下的49次都在等待。

接著使用多線程threading來改造程序：

import concurrent.futures

import requests

import threading

import time
thread_local = threading.local()
def get_session():

if not hasattr(thread_local, "session"):

thread_local.session = requests.Session()

return thread_local.session
def download_site(url):

session = get_session()

with session.get(url) as response:

print(f"下載了{len(response.content)}行數據")
def download_all_sites(sites):

with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:

executor.map(download_site, sites)
if __name__ == "__main__":
sites = ["https://v3u.cn"] * 50

start_time = time.time()

download_all_sites(sites)

duration = time.time() - start_time

print(f"下載了 {len(sites)}次，執行了{duration}秒")

這裡通過with關鍵詞開啟線程池上下文管理器，並發8個線程進行下載，程序返回：

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76161行數據

下載了76424行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了 50次，執行了7.680492877960205秒

很明顯，效率上有所提升，事實上，每個線程其實是在不停“切換”著運行，這就節省了單線程每次等待爬取結果的時間：

由此帶來了另外一個問題：上下文切換的時間開銷。

讓我們繼續改造，用協程來一試鋒芒，首先安裝異步web請求庫aiohttp:

pip3 install aiohttp

改寫邏輯：

import asyncio

import time

import aiohttp
async def download_site(session, url):

async with session.get(url) as response:

print(f"下載了{response.content_length}行數據")
async def download_all_sites(sites):

async with aiohttp.ClientSession() as session:

tasks = []

for url in sites:

task = asyncio.ensure_future(download_site(session, url))

tasks.append(task)

await asyncio.gather(*tasks, return_exceptions=True)
if __name__ == "__main__":

sites = ["https://v3u.cn"] * 50

start_time = time.time()

asyncio.run(download_all_sites(sites))

duration = time.time() - start_time

print(f"下載了 {len(sites)}次，執行了{duration}秒")

程序返回：


下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76424行數據

下載了76161行數據

下載了76424行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據

下載了76161行數據
下載了 50次，執行了6.893810987472534秒

效率上百尺竿頭更進一步，同樣的使用with關鍵字操作上下文管理器，協程使用asyncio.ensure_future()創建任務列表，該列表還負責啟動它們。創建所有任務後，使用asyncio.gather()來保持會話上下文的實例，直到所有爬取任務完成。和多線程threading的區別是，協程並不需要切換上下文，因此每個任務所需的資源和創建時間要少得多，因此創建和運行更多的任務效率更高：

綜上，並發邏輯歸根結底是減少CPU等待的時間，也就是讓CPU少等一會兒，而協程的工作方式顯然讓CPU等待的時間最少。

並行方式：多進程multiprocessing

再來試試多進程multiprocessing，並行能不能干並發的事？

import requests

import multiprocessing

import time
session = None
def set_global_session():

global session

if not session:

session = requests.Session()
def download_site(url):

with session.get(url) as response:

name = multiprocessing.current_process().name

print(f"讀了{len(response.content)}行")
def download_all_sites(sites):

with multiprocessing.Pool(initializer=set_global_session) as pool:

pool.map(download_site, sites)
if __name__ == "__main__":

sites = ["https://v3u.cn"] * 50

start_time = time.time()

download_all_sites(sites)

duration = time.time() - start_time

print(f"下載了 {len(sites)}次，執行了{duration}秒")

這裡我們依然使用上下文管理器開啟進程池，默認進程數匹配當前計算機的CPU核心數，也就是有幾核就開啟幾個進程，程序返回：

讀了76000行

讀了76241行

讀了76044行

讀了75894行

讀了76290行

讀了76312行

讀了76419行

讀了76753行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

讀了76290行

下載了 50次，執行了8.195281982421875秒

雖然比同步程序要快，但無疑的，效率上要低於多線程和協程。為什麼？因為多進程不適合IO密集型任務，雖然可以利用多核資源，但沒有任何意義：

無論開多少進程，CPU都沒有用武之地，多數情況下CPU都在等待IO操作，也就是說，多核反而拖累了IO程序的執行。

並行方式的選擇：CPU密集型任務

什麼是CPU密集型任務？這裡我們可以使用逆定理：所有不涉及硬盤讀寫(數據庫讀寫)、網絡請求、文件打印等任務都算CPU密集型任務任務，說白了就是，計算型任務。

以求平方和為例子：

import time
def cpu_bound(number):

return sum(i * i for i in range(number))
def find_sums(numbers):

for number in numbers:

cpu_bound(number)
if __name__ == "__main__":

numbers = [5_000_000 + x for x in range(20)]

start_time = time.time()

find_sums(numbers)

duration = time.time() - start_time

print(f"{duration}秒")

同步執行20次，需要花費多少時間？

4.466595888137817秒

再來試試並行方式：

import multiprocessing

import time
def cpu_bound(number):

return sum(i * i for i in range(number))
def find_sums(numbers):

with multiprocessing.Pool() as pool:

pool.map(cpu_bound, numbers)
if __name__ == "__main__":

numbers = [5_000_000 + x for x in range(20)]
start_time = time.time()

find_sums(numbers)

duration = time.time() - start_time

print(f"{duration}秒")

八核處理器，開八個進程開始跑：

1.1755797863006592秒

不言而喻，並行方式有效提高了計算效率。

最後，既然之前用並行方式運行了IO密集型任務，我們就再來試試用並發的方式運行CPU密集型任務：

import concurrent.futures

import time
def cpu_bound(number):

return sum(i * i for i in range(number))
def find_sums(numbers):

with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:

executor.map(cpu_bound, numbers)
if __name__ == "__main__":

numbers = [5_000_000 + x for x in range(20)]
start_time = time.time()

find_sums(numbers)

duration = time.time() - start_time

print(f"{duration}秒")

單進程開8個線程，走起：

4.452666759490967秒

如何？和並行方式運行IO密集型任務一樣，可以運行，但是沒有任何意義。為什麼？因為沒有任何IO操作了，CPU不需要等待了，CPU只要全力運算即可，所以你上多線程或者協程，無非就是畫蛇添足、多此一舉。

結語

有經驗的汽修師傅會告訴你，想省油就選CVT和雙離合，想質量穩定就選AT，經常高速上激烈駕駛就選雙離合，經常市區內堵車就選CVT；同樣地，作為經驗豐富的後台研發，你也可以告訴汽修師傅，任何不需要CPU等待的任務就選擇並行(multiprocessing)的處理方式，而需要CPU等待時間過長的任務，選擇並發(threading/asyncio)。反過來，我就想用CVT在高速上飙車，用雙離合在市區堵車，行不行？行，但沒有意義，或者說的更准確一些，沒有任何額外的收益；而用並發方式執行CPU密集型任務，用並行方式執行IO密集型任務行不行？也行，但依然沒有任何額外的收益，無他，唯物無定味，適口者珍矣。

原文轉載自「劉悅的技術博客」 https://v3u.cn/a_id_221

物無定味適口者珍，Python3並發場景(CPU密集/IO密集)任務的並發方式的場景抉擇(多線程threading/多進程multiprocessing/協程asyncio)的更多相關文章

【Python】【容器 | 迭代對象 | 迭代器 | 生成器 | 生成器表達式 | 協程 | 期物 | 任務】
Python 的 asyncio 類似於 C++ 的 Boost.Asio. 所謂「異步 IO」,就是你發起一個 IO 操作,卻不用等它結束,你可以繼續做其他事情,當它結束時,你會得到通知. Asyn ...
Python3 與 C# 並發編程之～協程篇
3.協程篇¶ 去年微信公眾號就陸陸續續發布了,我一直以為博客也匯總同步了,這幾天有朋友說一直沒找到,遂發現,的確是漏了,所以補上一篇在線預覽:https://github.lesschina.c ...
python3之協程
1.協程的概念協程,又稱微線程,纖程.英文名Coroutine. 線程是系統級別的它們由操作系統調度,而協程則是程序級別的由程序根據需要自己調度.在一個線程中會有很多函數,我們把這些函數稱為子程序, ...
11.python3標准庫--使用進程、線程和協程提供並發性
''' python提供了一些復雜的工具用於管理使用進程和線程的並發操作. 通過應用這些計數,使用這些模塊並發地運行作業的各個部分,即便是一些相當簡單的程序也可以更快的運行 subprocess提供了 ...
搞定python多線程和多進程
1 概念梳理: 1.1 線程 1.1.1 什麼是線程線程是操作系統能夠進行運算調度的最小單位.它被包含在進程之中,是進程中的實際運作單位.一條線程指的是進程中一個單一順序的控制流,一個進程中可以並發 ...
python3通過gevent.pool限制協程並發數量
協程雖然是輕量級的線程,但到達一定數量後,仍然會造成服務器崩潰出錯.最好的方法通過限制協程並發數量來解決此類問題. server代碼: #!/usr/bin/env python # -*- codi ...
python3 多線程和多進程
一.線程和進程 1.操作系統中,線程是CPU調度和分派的基本單位,線程依存於程序中 2.操作系統中,進程是系統進行資源分配和調度的一個基本單位,一個程序至少有一個進程 3.一個進程由至少一個線程組成, ...
std::list保存大量數據時，類型即是無析構函數，該list析構時會占用大量CPU
std::list保存大量數據時,類型即是無析構函數,該list析構時會占用大量CPU
python3下multiprocessing、threading和gevent性能對比----暨進程池、線程池和協程池性能對比
python3下multiprocessing.threading和gevent性能對比----暨進程池.線程池和協程池性能對比標簽: python3 / 線程池 / multiprocessi ...
轉搞定python多線程和多進程
轉自https://www.cnblogs.com/whatisfantasy/p/6440585.html 1 概念梳理: 1.1 線程 1.1.1 什麼是線程線程是操作系統能夠進行運算調度的最小 ...

隨機推薦

在C#中將String轉換成Enum：
一: 在C#中將String轉換成Enum: object Enum.Parse(System.Type enumType, string value, bool ignoreCase); 所以,我 ...
Debian下安裝vim
問題描述:安裝完系統以後,剛要打算開始寫程序,發現,vim還沒有裝,用su -切換到root後直接運行apt-get install vim,提示插入disc源,然後回車,陷入無法解決的狀態. 上網 ...
CStringUtf8ToUnicode
CString CStringUtf8ToUnicode( CString Utf8 ) { int wLen = 0; CString strUnicode; LPSTR pBufChar = NU ...
POJ 1797 Heavy Transportation (最短路)
Heavy Transportation Time Limit: 3000MS Memory Limit: 30000K Total Submissions: 22440 Accepted: ...
JS 獲取select(多選下拉)中所選值的示例代碼
通過js獲取select(多選下拉)中所選值,具體實現如下,有需要的朋友可以參考下,希望對大家有所幫助 <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML ...
UVA507-- Jill Rides Again
題目鏈接:http://uva.onlinejudge.org/index.php?option=com_onlinejudge&Itemid=8&page=show_problem& ...
貼片晶振跟隨WIFI滲透到我們的生活之中
在用WIFI上網的時候很多時候會遇到上網速度不是慢就是卡,甚至是連接不上更氣人的就是有時候會識別不了.我最常遇到的就是連接不上去,玩著玩著就玩不了了.起初我不知道這是什麼原因,但是從事晶振行業之後我慢 ...
[HNOI2012]與非
題目描述 NAND(與非)是一種二元邏輯運算,其運算結果為真當且僅當兩個輸入的布爾值不全為真.NAND運算的真值表如下(1表示真,0表示假): 兩個非負整數的NAND是指將它們表示成二進制數,再在對應 ...
Chrome DevTools: Export your raw Code Coverage Data
The Code Coverage tool visually shows you which lines of code in your CSS and JavaScript are used an ...
MySql使用存儲過程清除數據庫所有表數據，保存數據結構
BEGIN DECLARE strClear VARCHAR(256); DECLARE done INT DEFAULT 0; #定義游標 DECLARE curOne CURSOR FOR sel ...