關(guān)于什么是并發(fā)模型，我在這里引用 Go 語(yǔ)言聯(lián)合創(chuàng)造者 Rob Pike 的一段話(huà)：

并發(fā)是指一次處理多件事。并行是指一次做多件事。二者不同，但是有聯(lián)系。一個(gè)關(guān)于結(jié)構(gòu)，一個(gè)關(guān)于執(zhí)行。并發(fā)用于制定方案，用來(lái)解決可能（但未必）并行的問(wèn)題。

在不涉及并發(fā)概念的情況下，一個(gè)單進(jìn)程單線(xiàn)程的程序執(zhí)行情況可能是這樣的：調(diào)用一個(gè)函數(shù)，發(fā)出調(diào)用的代碼開(kāi)始等待函數(shù)執(zhí)行完成，直到函數(shù)返回結(jié)果，如果函數(shù)拋出異常，則可以把調(diào)用函數(shù)的代碼放到 try/except 語(yǔ)句塊中，來(lái)捕獲和處理異常。

但是，當(dāng)涉及到并發(fā)時(shí)，情況就沒(méi)這么簡(jiǎn)單了。在啟用多線(xiàn)程（或多進(jìn)程）后，你無(wú)法在一個(gè)線(xiàn)程（或進(jìn)程）中知道另一個(gè)線(xiàn)程（或進(jìn)程）被調(diào)用的函數(shù)何時(shí)執(zhí)行完成，也無(wú)法輕松得知函數(shù)調(diào)用結(jié)果或捕獲異常。只能采用某種通知的方式，來(lái)進(jìn)行線(xiàn)程（或進(jìn)程）間通信，這可能是一個(gè)信號(hào)，也可能是一個(gè)消息隊(duì)列等。

本文主要講解如何讓 Python 能夠同時(shí)處理多個(gè)任務(wù)，即如何使用并發(fā)模型編程。

目標(biāo)

我們將要實(shí)現(xiàn)一個(gè)旋轉(zhuǎn)指針程序，啟動(dòng)一個(gè)程序，阻塞 3 秒鐘（模擬耗時(shí)任務(wù)），在這期間，終端展示字符動(dòng)畫(huà)，讓用戶(hù)知道程序仍在執(zhí)行，并沒(méi)有停止，3 秒后程序打印耗時(shí)任務(wù)的計(jì)算結(jié)果并退出。

實(shí)現(xiàn)好的程序效果如下：

這有點(diǎn)像下載進(jìn)度條，因?yàn)榇蛴⌒D(zhuǎn)指針和耗時(shí)任務(wù)是“同時(shí)”進(jìn)行的，這種場(chǎng)景只能通過(guò)并發(fā)模型來(lái)實(shí)現(xiàn)。

我們將分別使用多線(xiàn)程、多進(jìn)程以及協(xié)程來(lái)實(shí)現(xiàn)這個(gè)程序，以此來(lái)演示 Python 的的并發(fā)模型用法。

多線(xiàn)程版

第一版旋轉(zhuǎn)指針程序使用 Python 多線(xiàn)程來(lái)編寫(xiě)，首先，我們需要定義兩個(gè)函數(shù) spin、slow 分別用來(lái)實(shí)現(xiàn)旋轉(zhuǎn)指針和模擬耗時(shí)任務(wù)（比如從網(wǎng)上下載一個(gè)文件）。

import itertools
import time
from threading import Thread, Event
def spin(msg: str, done: Event) -> None:
    for char in itertools.cycle(r'\|/-'):
        status = f'\r{char} {msg}'
        print(status, end='', flush=True)
        if done.wait(0.1):
            break
    blanks = ' ' * len(status)
    print(f'\r{blanks}\r', end='')
def slow() -> int:
    time.sleep(3)
    return 42

threading 模塊提供多線(xiàn)程支持，Thread 實(shí)例用來(lái)管理一個(gè)新的線(xiàn)程，Event 可以用來(lái)進(jìn)行線(xiàn)程間通信。

spin 函數(shù)將作為一個(gè)任務(wù)在單獨(dú)的線(xiàn)程中執(zhí)行，它接收兩個(gè)參數(shù) msg、done，傳遞進(jìn)來(lái)的 msg 將會(huì)跟隨旋轉(zhuǎn)指針一起被打印，done 參數(shù)類(lèi)型為 threading.Event，用來(lái)實(shí)現(xiàn)多個(gè)線(xiàn)程間的通信，以此來(lái)同步任務(wù)狀態(tài)。

itertools.cycle(r'\|/-') 是一個(gè)無(wú)限迭代器，一次產(chǎn)出一個(gè)字符，不停的迭代。比如用 for 遍歷 itertools.cycle('123')，將得到無(wú)限迭代的數(shù)據(jù) 123123123...。這里 \|/- 字符不停迭代并被打印，就會(huì)產(chǎn)生旋轉(zhuǎn)指針的效果。

打印的 status 字符串第一個(gè)字符為 \r，可以實(shí)現(xiàn)將光標(biāo)移動(dòng)到行首，這是一個(gè)使用文本在控制臺(tái)實(shí)現(xiàn)動(dòng)畫(huà)的小技巧。

接下來(lái)的 done.wait(0.1) 是這個(gè)函數(shù)的關(guān)鍵代碼，它是主線(xiàn)程與執(zhí)行當(dāng)前函數(shù)的子線(xiàn)程之間通信的橋梁。done.wait 方法簽名為 Event.wait(self, timeout=None)，該方法等待 timeout 指定的時(shí)間后返回 False，我們?cè)谶@里指定為 0.1 秒。如果在其他線(xiàn)程中使用 Event.set() 設(shè)置了這個(gè)事件，則當(dāng)前線(xiàn)程該方法將立即返回 True，此時(shí) for 循環(huán)就會(huì)被 break 掉。

spin 函數(shù)在退出前，還會(huì)打印幾個(gè)空格來(lái)實(shí)現(xiàn)清空當(dāng)前行打印內(nèi)容的效果，并且最終還將光標(biāo)移動(dòng)到行首。

slow 函數(shù)使用 time.sleep(3) 暫停 3 秒，模擬耗時(shí)操作，這個(gè)函數(shù)將像我們往常編寫(xiě)的單線(xiàn)程代碼一樣在主線(xiàn)程中執(zhí)行。

接下來(lái)我們要編寫(xiě)多線(xiàn)程代碼來(lái)分別調(diào)用 spin 和 slow 兩個(gè)函數(shù)，完成這個(gè)旋轉(zhuǎn)指針程序。

def supervisor() -> int:
    done = Event()
    spinner = Thread(target=spin, args=('thinking!', done))
    print(f'spinner object: {spinner}')
    spinner.start()
    result = slow()
    done.set()
    spinner.join()
    return result
def main() -> None:
    result = supervisor()
    print(f'Answer: {result}')
if __name__ == '__main__':
    main()

supervisor 函數(shù)中，首先實(shí)例化了一個(gè) Event 對(duì)象，用于多線(xiàn)程通信。

接著，又實(shí)例化了一個(gè) Thread 對(duì)象，用來(lái)管理子線(xiàn)程，target 參數(shù)接收一個(gè)函數(shù) spin，這個(gè)函數(shù)將在一個(gè)獨(dú)立的子線(xiàn)程中執(zhí)行，args 參數(shù)接收一個(gè)元組，在子線(xiàn)程中調(diào)用 spin 函數(shù)時(shí)，元組的各個(gè)參數(shù)將被原樣傳遞給 spin 函數(shù)。

Thread 對(duì)象必須要顯式的調(diào)用 start 方法才能啟動(dòng)，所以代碼執(zhí)行到 spinner.start() 時(shí)，子線(xiàn)程才會(huì)真正開(kāi)始執(zhí)行。子線(xiàn)程只會(huì)執(zhí)行 spin 函數(shù)，至于下方的代碼與子線(xiàn)程無(wú)關(guān)，都是主線(xiàn)程要執(zhí)行的代碼。

子線(xiàn)程的執(zhí)行對(duì)主線(xiàn)程執(zhí)行不會(huì)產(chǎn)生影響，主線(xiàn)程代碼會(huì)繼續(xù)往下運(yùn)行，主線(xiàn)程調(diào)用 slow() 時(shí)會(huì)被耗時(shí)任務(wù)所阻塞。此時(shí)，子線(xiàn)程內(nèi)部代碼執(zhí)行不受影響，所以子線(xiàn)程會(huì)不停的打印旋轉(zhuǎn)指針。

等待 3 秒結(jié)束后，主線(xiàn)程中 slow() 函數(shù)返回結(jié)果，主線(xiàn)程調(diào)用 done.set() 將 Event 對(duì)象設(shè)置為 True。此時(shí)，子線(xiàn)程 spin 函數(shù)內(nèi)部 done.wait(0.1) 會(huì)立即返回 True，隨即 for循環(huán)終止，spin 執(zhí)行完成后子線(xiàn)程也就退出了。

主線(xiàn)程不受子線(xiàn)程退出影響，會(huì)接著往下執(zhí)行，調(diào)用 spinner.join() 是為了等待子線(xiàn)程結(jié)束，主線(xiàn)程會(huì)阻塞在這里，保證子線(xiàn)程結(jié)束后才會(huì)往下執(zhí)行。顯然，子線(xiàn)程在執(zhí)行完 spin 函數(shù)就結(jié)束了，所以主線(xiàn)程代碼會(huì)繼續(xù)往下執(zhí)行。

supervisor 函數(shù)最終返回 slow 方法的返回值 result。

入口函數(shù) main 打印 result 值后，主線(xiàn)程也退出了，程序終止。

以上，就是多線(xiàn)程版旋轉(zhuǎn)指針程序的全部邏輯了。

我們來(lái)測(cè)試下這個(gè)程序執(zhí)行效果：

多線(xiàn)程對(duì)象 spinner 輸出結(jié)果為 <Thread(Thread-1, initial)>，其中 Thread-1 是線(xiàn)程名稱(chēng)，initial 是線(xiàn)程狀態(tài)，表示當(dāng)前線(xiàn)程剛初始化完成，尚未啟動(dòng)。

多進(jìn)程版

Python 提供了 multiprocessing 來(lái)支持多進(jìn)程，這個(gè)模塊的 API 基本模仿了多線(xiàn)程的 threading 模塊，所以有了前文的基礎(chǔ)，多進(jìn)程代碼也非常容易看懂。

同多線(xiàn)程一樣，multiprocessing 包也為多進(jìn)程通信提供了 Event 對(duì)象。不同的是，threading.Event 是一個(gè)類(lèi)，multiprocessing.Event 是一個(gè)函數(shù)，它返回一個(gè) synchronize.Event 類(lèi)實(shí)例。所以 spin 函數(shù)簽名需要進(jìn)行如下修改：

from multiprocessing import Process, Event
from multiprocessing import synchronize
def spin(msg: str, done: synchronize.Event) -> None:
    ...

spin 函數(shù)內(nèi)部代碼無(wú)需調(diào)整，只需要修改參數(shù) done 的類(lèi)型注解即可。所以不難發(fā)現(xiàn) synchronize.Event 同樣支持 Event.wait(self, timeout=None) 方法。

多進(jìn)程版本的 supervisor 函數(shù)也要稍作修改：

def supervisor() -> int:
    done = Event()
    spinner = Process(target=spin, args=('thinking!', done))
    print(f'spinner object: {spinner}')
    spinner.start()
    result = slow()
    done.set()
    spinner.join()
    return result

雖然 multiprocessing.Event 和 threading.Event 類(lèi)型不同，但二者用法和作用則完全相同。

這里使用 Process 實(shí)例化一個(gè)進(jìn)程對(duì)象，Process 用法和 Thread 用法同樣如出一轍。

只需要對(duì)代碼做少量的改動(dòng)，我們就將程序從多線(xiàn)程遷移到了多進(jìn)程。這一點(diǎn)，Python 做的非常友好，掌握了多線(xiàn)程編程，基本上就掌握了多進(jìn)程編程，我們只需要在適當(dāng)?shù)臅r(shí)候，使用不同的模塊即可。

下面是多進(jìn)程版本旋轉(zhuǎn)指針程序測(cè)試效果：

多進(jìn)程對(duì)象 spinner 輸出結(jié)果為 <Process name='Process-1' parent=94367 initial>，進(jìn)程名稱(chēng)為 Process-1，parent 代表父進(jìn)程 ID 為 94367（即主進(jìn)程 ID），initial 是進(jìn)程狀態(tài)，表示當(dāng)前進(jìn)程剛初始化完成，尚未啟動(dòng)。

協(xié)程版

最后我們將使用協(xié)程實(shí)現(xiàn)旋轉(zhuǎn)指針程序，這一版本代碼改動(dòng)會(huì)比較大。

Python 在 3.5 版本提供了 async、await 關(guān)鍵字（可以參考 PEP 492），開(kāi)始原生支持了協(xié)程，我們不再需要編寫(xiě)難懂的 yeild from 來(lái)使用生成器實(shí)現(xiàn)協(xié)程功能了。

Python 協(xié)程通常在單線(xiàn)程的事件循環(huán)中運(yùn)行。協(xié)程是一個(gè)可以?huà)炱鹱陨聿⒃谝院蠡謴?fù)的“函數(shù)”，async 用來(lái)定義協(xié)程，一個(gè)協(xié)程必須顯式的使用 await 關(guān)鍵字主動(dòng)讓出控制權(quán)，另一個(gè)協(xié)程才有機(jī)會(huì)在主事件循環(huán)的調(diào)度下并發(fā)的執(zhí)行。

協(xié)程版本旋轉(zhuǎn)指針程序需要對(duì) spin 和 slow 兩個(gè)函數(shù)做如下修改：

import asyncio
import itertools
async def spin(msg: str) -> None:
    for char in itertools.cycle(r'\|/-'):
        status = f'\r{char} {msg}'
        print(status, end='', flush=True)
        try:
            await asyncio.sleep(0.1)
        except asyncio.CancelledError:
            break
    blanks = ' ' * len(status)
    print(f'\r{blanks}\r', end='')
async def slow() -> int:
    await asyncio.sleep(3)
    return 42

首先我們使用 async def 將 spin 定義為一個(gè)協(xié)程，讓其不再是一個(gè)常規(guī)的函數(shù)。

spin 協(xié)程取消了第二個(gè)參數(shù)，因?yàn)?Python 沒(méi)有為協(xié)程提供 Event 對(duì)象來(lái)進(jìn)行通信，我們需要采用其他招式。

在原來(lái)使用 Event 通信的地方替換成了由 try/except 包裹的 await asyncio.sleep(0.1) 語(yǔ)句塊代碼。這段代碼塊有如下三個(gè)作用：

• await asyncio.sleep(0.1) 的作用類(lèi)似 time.sleep，可以讓程序暫停 0.1 秒。不同的是，使用 await asyncio.sleep 暫停時(shí)不阻塞其他協(xié)程。
• 因?yàn)檫@里加入了 await 關(guān)鍵字，代碼執(zhí)行到這里時(shí)，當(dāng)前協(xié)程會(huì)主動(dòng)讓出控制權(quán)，不再繼續(xù)往下執(zhí)行，由事件循環(huán)來(lái)調(diào)度其他協(xié)程執(zhí)行。
• 如果在控制當(dāng)前協(xié)程的 Task 實(shí)例中調(diào)用 cancel 方法（有關(guān) Task 的內(nèi)容稍后會(huì)進(jìn)行講解），await asyncio.sleep(0.1) 會(huì)拋出 CancelledError 異常，這里使用 try/except 捕獲異常后退出循環(huán)。這樣，我們就在多個(gè)協(xié)程間利用異常機(jī)制完成了通信，而不必借助于額外的 Event 對(duì)象。

slow 函數(shù)也被改造為一個(gè)協(xié)程，其內(nèi)部原來(lái)編寫(xiě)的阻塞代碼 time.sleep(3) 被替換為了 await asyncio.sleep(3)。

可以發(fā)現(xiàn)，其實(shí)協(xié)程與普通的函數(shù)在定義上差別不大，只不過(guò)多了兩個(gè)關(guān)鍵字 async 和 await。但二者在執(zhí)行方式上大有不同，普通函數(shù)在使用 () 運(yùn)算符調(diào)用時(shí)（即 spin()）會(huì)立刻執(zhí)行，而協(xié)程在使用 spin() 時(shí)只會(huì)創(chuàng)建一個(gè)協(xié)程對(duì)象，不會(huì)執(zhí)行。

要執(zhí)行上面兩個(gè)協(xié)程對(duì)象，我們還要對(duì) supervisor 和 main 函數(shù)進(jìn)行改造：

async def supervisor() -> int:
    spinner = asyncio.create_task(spin('thinking!'))
    print(f'spinner object: {spinner}')
    result = await slow()
    spinner.cancel()
    return result
def main() -> None:
    result = asyncio.run(supervisor())
    print(f'Answer: {result}')
if __name__ == '__main__':
    main()

supervisor 函數(shù)同樣被修改為協(xié)程，spin('thinking!') 并不會(huì)像函數(shù)一樣立即執(zhí)行，只會(huì)創(chuàng)建一個(gè)協(xié)程對(duì)象，將它傳遞給 asyncio.create_task，我們可以得到一個(gè) asyncio.Task 對(duì)象，這個(gè) Task 對(duì)象包裝了協(xié)程對(duì)象并調(diào)度其執(zhí)行，它還提供控制和查詢(xún)協(xié)程對(duì)象運(yùn)行狀態(tài)的方法。

使用 await 關(guān)鍵字來(lái)調(diào)用 slow 協(xié)程，這將阻塞 supervisor 程序（但會(huì)讓出控制權(quán)，使其他協(xié)程得以執(zhí)行），直到 slow 返回，返回結(jié)果賦值給 result 變量。

接著調(diào)用了 spinner.cancel()，Task.cancel 方法調(diào)用后，將立即在 Task 所包裝的協(xié)程對(duì)象即 spin 協(xié)程中拋出 CancelledError 異常，spin 中需要使用 try/except 捕獲 await asyncio.sleep(0.1) 拋出的異常，這樣，就實(shí)現(xiàn)了不同協(xié)程之間通過(guò)異常進(jìn)行通信。

main 是唯一的普通函數(shù)，沒(méi)有被改造為協(xié)程。main 函數(shù)中的 asyncio.run 是整個(gè)協(xié)程的啟動(dòng)入口，asyncio.run 函數(shù)啟動(dòng)事件循環(huán)，驅(qū)動(dòng) supervisor() 協(xié)程運(yùn)行，最終也將啟動(dòng)其他協(xié)程。

在以上示例代碼中，我們可以總結(jié)出運(yùn)行協(xié)程的 3 種方式：

• asyncio.run(coroutine())：在一個(gè)常規(guī)函數(shù)中調(diào)用，是協(xié)程啟動(dòng)入口，將開(kāi)啟協(xié)程的事件循環(huán)，調(diào)用后保持阻塞，直至拿到 coroutine() 的返回結(jié)果。
• asyncio.create_task(coroutine())：在協(xié)程中調(diào)用，接收另一個(gè)協(xié)程對(duì)象并調(diào)度其最終執(zhí)行，返回的 Task 對(duì)象是對(duì)協(xié)程對(duì)象的包裝，并且提供控制和查詢(xún)協(xié)程對(duì)象運(yùn)行狀態(tài)的方法。
• await coroutine()：在協(xié)程中調(diào)用，await 關(guān)鍵字主動(dòng)讓出執(zhí)行控制權(quán)，終止當(dāng)前協(xié)程執(zhí)行，直至拿到 coroutine() 的返回結(jié)果。同時(shí)這也是一個(gè)表達(dá)式，返回結(jié)果即為 coroutine() 返回值。

下面是協(xié)程版本旋轉(zhuǎn)指針程序測(cè)試效果：

在協(xié)程版本中，spinner 是一個(gè) Task 對(duì)象，其字符串表示形式為 <Task pending name='Task-2' coro=<spin() running at /Users/jianghushinian/spin/spinner_async.py:8>>。

根據(jù)以上示例代碼，我們可以總結(jié)出 Python 協(xié)程的最大特點(diǎn)：一處異步，處處異步。在協(xié)程中任何耗時(shí)操作都會(huì)減慢事件循環(huán)，由于事件循環(huán)是單線(xiàn)程管理的，所以這會(huì)影響其他所有協(xié)程。在編寫(xiě)協(xié)程代碼，要格外小心，不要寫(xiě)出同步阻塞的代碼。好在如今 Python 已經(jīng)從語(yǔ)法層面原生支持協(xié)程，比使用生成器實(shí)現(xiàn)協(xié)程的年代要好多了。

給你留個(gè)小作業(yè)：嘗試將 slow 協(xié)程中 await asyncio.sleep(3) 替換成普通的 time.sleep(3) 觀察下效果并思考為什么。

總結(jié)

本文我們分別使用了多線(xiàn)程、多進(jìn)程以及協(xié)程這三種不同的并發(fā)模型實(shí)現(xiàn)了旋轉(zhuǎn)指針程序。三者比較起來(lái)，多線(xiàn)程、多進(jìn)程在語(yǔ)法上差別不大，協(xié)程則大為不同，理解起來(lái)也更加困難。

由 supervisor 中打印的 spinner 對(duì)象結(jié)果可以看出，線(xiàn)程對(duì)象使用 Thread 來(lái)表示，進(jìn)程對(duì)象使用 Process 來(lái)表示，協(xié)程對(duì)象則使用 Task 來(lái)表示。協(xié)程的定位是用戶(hù)態(tài)線(xiàn)程，相比傳統(tǒng)意義上的線(xiàn)程更加輕量，所以叫作 Task 也合理，代表同一個(gè)線(xiàn)程下的不同任務(wù)。

線(xiàn)程和進(jìn)程無(wú)法在外部終止，即主線(xiàn)程和主進(jìn)程無(wú)法終止由子線(xiàn)程或子進(jìn)程來(lái)執(zhí)行的 spin 函數(shù)，只能通過(guò) Event 來(lái)進(jìn)行通信，然后由 spin 函數(shù)內(nèi)部自己終止。協(xié)程則可以通過(guò)任務(wù)實(shí)例方法 Task.cancel() 進(jìn)行通信，spin 協(xié)程中捕獲 CancelledError 異常后終止自身代碼。

記住，使用協(xié)程的代碼只有一個(gè)執(zhí)行流，就如同單線(xiàn)程代碼，同樣只有一個(gè)執(zhí)行流，只不過(guò)單線(xiàn)程代碼執(zhí)行流永遠(yuǎn)是從上到下，而協(xié)程的執(zhí)行流則由事件循環(huán)來(lái)控制。

多線(xiàn)程和多進(jìn)程模型是搶占式的，由操作系統(tǒng)進(jìn)行調(diào)度執(zhí)行。用戶(hù)通常需要控制的是不要讓多個(gè)線(xiàn)程（進(jìn)程）同時(shí)操作同一個(gè)數(shù)據(jù)，常使用互斥鎖來(lái)解決這一問(wèn)題。而協(xié)程只有一個(gè)控制循環(huán)，協(xié)程的控制權(quán)在我們自己手里，我們決定什么時(shí)候切換其他任務(wù)來(lái)執(zhí)行。所以在編寫(xiě)協(xié)程代碼時(shí)，要時(shí)刻注意不要寫(xiě)出同步阻塞代碼。

到此這篇關(guān)于詳解Python如何使用并發(fā)模型編程的文章就介紹到這了,更多相關(guān)Python并發(fā)模型內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論