Python中實現(xiàn)多線程 Threading 和多進(jìn)程 Multiprocessing
本文轉(zhuǎn)載自微信公眾號「Piper蛋窩」，作者Piper蛋 。轉(zhuǎn)載本文請聯(lián)系Piper蛋窩公眾號。
昨天晚上組會輪到我匯報技術(shù)內(nèi)容，最近正在和 ray 以及 spark 打交道，索性講一下并發(fā)和并行。反正大家都是管理學(xué)院的，平時很少接觸這種，因此這個選題不大可能因為內(nèi)容基礎(chǔ)而貽笑大方。
本文擺一擺并發(fā)和并行。附上很簡單的 Python 代碼，涉及到自帶庫threading[1] 和 multiprocessing[2] 的使用。
并發(fā)和并行
咱們簡單用多線程對應(yīng)并發(fā)，多進(jìn)程對應(yīng)并行。多線程并發(fā)更強(qiáng)調(diào)充分利用性能;多進(jìn)程并行更強(qiáng)調(diào)提升性能上限。
我用非常簡單且不那么嚴(yán)謹(jǐn)?shù)谋扔鱽碚f明。
多線程
一個 CPU 相當(dāng)于一個學(xué)生。
一個學(xué)生一周開一次組會，換句話說一周給老師匯報一次工作。
老師一般會給學(xué)生同時布置幾個任務(wù)，比如做比賽、做項目、讀論文，學(xué)生可能周一做做比賽、周二讀讀論文、周三做做項目... 到了組會，他就把三件事都拿出來匯報，老師很欣慰，因為在老師的視角里：學(xué)生這三件事是同時在做的。
多線程也是同一個道理，假設(shè)你的手機(jī)只有一塊單核 CPU 。你的 CPU 這 0.01 秒用來播放音樂，下 0.01 秒用來解析網(wǎng)頁... 在你的視角里：播放音樂和解析網(wǎng)頁是同時進(jìn)行的。你大可以暢快地邊聽音樂邊網(wǎng)上沖浪
何謂充分利用性能? 如果這學(xué)生只有一項工作，那他這一周可能只需要花費兩天來做任務(wù)，剩下時間摸魚(針不搓，三點鐘飲茶先!)。因此，我們用「多線程」來讓學(xué)生實現(xiàn)『并發(fā)』，充分利用學(xué)生能力。
在實際情況中，多線程、高并發(fā)這些詞語更多地出現(xiàn)在服務(wù)端程序里。比如一個網(wǎng)絡(luò)連接由一個線程負(fù)責(zé)，一塊 CPU 可以負(fù)責(zé)處理多個異步的請求，大大提升了 CPU 利用率。
多進(jìn)程
多個 CPU ( CPU 的多核)相當(dāng)于多個學(xué)生。
一個任務(wù)可以拆成幾個任務(wù)相互協(xié)作、同時進(jìn)行，則是多進(jìn)程。
比如研究生課程，老師非得留個論文作業(yè)，都研究生了我去，留啥大作業(yè)。
那咱就多線程并行搞唄。確定了大概思路，剩下的一股腦寫就行。咱隊伍里一共甲乙丙丁四名同學(xué)，那就：
- 甲同學(xué)負(fù)責(zé) Introduction
- 乙同學(xué)負(fù)責(zé) Background
- 丙同學(xué)負(fù)責(zé) Related Works
- 丁同學(xué)負(fù)責(zé) Methodology
這是乙同學(xué)提出異議：不應(yīng)該是先完成 Introduction 再寫 Background ，一個個來嘛?
大哥，都研究生了嗷，作業(yè)糊弄糊弄差不多得了啊。讓你寫你就寫。
可以預(yù)知，上述四部分同時進(jìn)行，怎么也比一個人寫四塊要快。
所以說 多進(jìn)程并行提升性能上限 。
在實際情況中，多進(jìn)程更多地與高性能計算、分布式計算聯(lián)系在一起。
Python 實現(xiàn)
首先聲明咱的實驗環(huán)境。
- > python --version
- Python 3.8.5
咱們設(shè)置個任務(wù)：求數(shù)的歐拉函數(shù)值。
- def euler_func(n: int) -> int:
- res = n
- i = 2
- while i <= n // i:
- if n % i == 0:
- res = res // i * (i - 1)
- while (n % i == 0): n = n // i
- i += 1
- if n > 1:
- res = res // n * (n - 1)
- return res
求一個數(shù)的歐拉函數(shù)值可能很快，但是一堆數(shù)呢?
所以咱想著用并行完成這個任務(wù)。
咱們把任務(wù)分成三份。
- task1 = list(range(2, 50000, 3)) # 2, 5, ...
- task2 = list(range(3, 50000, 3)) # 3, 6, ...
- task3 = list(range(4, 50000, 3)) # 4, 7, ...
- def job(task: List):
- for t in task:
- euler_func(t)
來看看平平無奇的正常串行。
- @timer
- def normal():
- job(task1)
- job(task2)
- job(task3)
完成了 task1 再完成 task2 ... 行，沒毛病。
看看多線程?
- import threading as th
- @timer
- def mutlthread():
- th1 = th.Thread(target=job, args=(task1, ))
- th2 = th.Thread(target=job, args=(task2, ))
- th3 = th.Thread(target=job, args=(task3, ))
- th1.start()
- th2.start()
- th3.start()
- th1.join()
- th2.join()
- th3.join()
再看看多進(jìn)程?
- import multiprocessing as mp
- @timer
- def multcore():
- p1 = mp.Process(target=job, args=(task1, ))
- p2 = mp.Process(target=job, args=(task2, ))
- p3 = mp.Process(target=job, args=(task3, ))
- p1.start()
- p2.start()
- p3.start()
- p1.join()
- p2.join()
- p3.join()
上述代碼的邏輯是這樣的：
- 我創(chuàng)建線程/進(jìn)程，其生來的目的就是完成任務(wù)job(task1)或job(task2)、job(task3)，注意這里函數(shù)名和參數(shù)被分開了target=job, args=(task1, )
- 然后 start() ，告訴線程/進(jìn)程：你可以開始干活了
- 他們自己干自己的，咱們程序主邏輯還得繼續(xù)往下運行
- 到 join() 這里，咱們是指讓線程/進(jìn)程阻塞住咱的主邏輯，比如p1.join()是指：p1不干完活，我主邏輯不往下進(jìn)行(屬于是「阻塞」)
- 這樣，我們的函數(shù)multcore結(jié)束后，一定其中的線程/進(jìn)程任務(wù)都完成了
咱看看結(jié)果：
- if __name__ == '__main__':
- print("同步串行：")
- normal()
- print("多線程并發(fā)：")
- mutlthread()
- print("多進(jìn)程并行：")
- multcore()
- # 下面是結(jié)果
- 同步串行：
- timer: using 0.24116 s
- 多線程并發(fā)：
- timer: using 0.24688 s
- 多進(jìn)程并行：
- timer: using 0.13791 s
結(jié)果不太對，按理說，多進(jìn)程并行的耗時應(yīng)該是同步串行的三分之一，畢竟三個同等體量的任務(wù)在同時進(jìn)行。
多線程并發(fā)比同步串行慢是應(yīng)該的，因為多線程并發(fā)和同步串行的算力是一樣的，但是多線程并發(fā)得在各個任務(wù)間來回切換，導(dǎo)致更慢。
你問 @timer 是什么意思?哦，這個是我寫的修飾器，如下。
- def timer(func):
- @wraps(func)
- def inner_func():
- t = time.time()
- rts = func()
- print(f"timer: using {time.time() - t :.5f} s")
- return rts
- return inner_func
不太明白『Python修飾器』的老鐵，不如給我點個「在看」，再關(guān)注下我，咱們以后詳細(xì)道來。
