[[405163]]

本文轉(zhuǎn)載自微信公眾號(hào)「Piper蛋窩」，作者Piper蛋 。轉(zhuǎn)載本文請聯(lián)系Piper蛋窩公眾號(hào)。

昨天晚上組會(huì)輪到我匯報(bào)技術(shù)內(nèi)容，最近正在和 ray 以及 spark 打交道，索性講一下并發(fā)和并行。反正大家都是管理學(xué)院的，平時(shí)很少接觸這種，因此這個(gè)選題不大可能因?yàn)閮?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明。

多線程

一個(gè) CPU 相當(dāng)于一個(gè)學(xué)生。

一個(gè)學(xué)生一周開一次組會(huì)，換句話說一周給老師匯報(bào)一次工作。

老師一般會(huì)給學(xué)生同時(shí)布置幾個(gè)任務(wù)，比如做比賽、做項(xiàng)目、讀論文，學(xué)生可能周一做做比賽、周二讀讀論文、周三做做項(xiàng)目... 到了組會(huì)，他就把三件事都拿出來匯報(bào)，老師很欣慰，因?yàn)樵诶蠋煹囊暯抢铮簩W(xué)生這三件事是同時(shí)在做的。

多線程也是同一個(gè)道理，假設(shè)你的手機(jī)只有一塊單核 CPU 。你的 CPU 這 0.01 秒用來播放音樂，下 0.01 秒用來解析網(wǎng)頁... 在你的視角里：播放音樂和解析網(wǎng)頁是同時(shí)進(jìn)行的。你大可以暢快地邊聽音樂邊網(wǎng)上沖浪

何謂充分利用性能? 如果這學(xué)生只有一項(xiàng)工作，那他這一周可能只需要花費(fèi)兩天來做任務(wù)，剩下時(shí)間摸魚(針不搓，三點(diǎn)鐘飲茶先!)。因此，我們用「多線程」來讓學(xué)生實(shí)現(xiàn)『并發(fā)』，充分利用學(xué)生能力。

在實(shí)際情況中，多線程、高并發(fā)這些詞語更多地出現(xiàn)在服務(wù)端程序里。比如一個(gè)網(wǎng)絡(luò)連接由一個(gè)線程負(fù)責(zé)，一塊 CPU 可以負(fù)責(zé)處理多個(gè)異步的請求，大大提升了 CPU 利用率。

多進(jìn)程

多個(gè) CPU ( CPU 的多核)相當(dāng)于多個(gè)學(xué)生。

一個(gè)任務(wù)可以拆成幾個(gè)任務(wù)相互協(xié)作、同時(shí)進(jìn)行，則是多進(jìn)程。

比如研究生課程，老師非得留個(gè)論文作業(yè)，都研究生了我去，留啥大作業(yè)。

那咱就多線程并行搞唄。確定了大概思路，剩下的一股腦寫就行。咱隊(duì)伍里一共甲乙丙丁四名同學(xué)，那就：

• 甲同學(xué)負(fù)責(zé) Introduction
• 乙同學(xué)負(fù)責(zé) Background
• 丙同學(xué)負(fù)責(zé) Related Works
• 丁同學(xué)負(fù)責(zé) Methodology

這是乙同學(xué)提出異議：不應(yīng)該是先完成 Introduction 再寫 Background ，一個(gè)個(gè)來嘛?

大哥，都研究生了嗷，作業(yè)糊弄糊弄差不多得了啊。讓你寫你就寫。

可以預(yù)知，上述四部分同時(shí)進(jìn)行，怎么也比一個(gè)人寫四塊要快。

所以說 多進(jìn)程并行提升性能上限 。

在實(shí)際情況中，多進(jìn)程更多地與高性能計(jì)算、分布式計(jì)算聯(lián)系在一起。

Python 實(shí)現(xiàn)

首先聲明咱的實(shí)驗(yàn)環(huán)境。

> python --version 
Python 3.8.5 

咱們設(shè)置個(gè)任務(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 

求一個(gè)數(shù)的歐拉函數(shù)值可能很快，但是一堆數(shù)呢?

所以咱想著用并行完成這個(gè)任務(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ù)往下運(yùn)行
• 到 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)程并行的耗時(shí)應(yīng)該是同步串行的三分之一，畢竟三個(gè)同等體量的任務(wù)在同時(shí)進(jìn)行。

多線程并發(fā)比同步串行慢是應(yīng)該的，因?yàn)槎嗑€程并發(fā)和同步串行的算力是一樣的，但是多線程并發(fā)得在各個(gè)任務(wù)間來回切換，導(dǎo)致更慢。

你問 @timer 是什么意思?哦，這個(gè)是我寫的修飾器，如下。

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修飾器』的老鐵，不如給我點(diǎn)個(gè)「在看」，再關(guān)注下我，咱們以后詳細(xì)道來。