快捷導(dǎo)航

Python進(jìn)階之協(xié)程詳解

更新時(shí)間：2022年01月18日 17:27:43 作者：云物互聯(lián)

這篇文章主要為大家介紹了Python進(jìn)階之協(xié)程，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來(lái)幫助

協(xié)程

協(xié)程（co-routine，又稱微線程）是一種多方協(xié)同的工作方式。當(dāng)前執(zhí)行者在某個(gè)時(shí)刻主動(dòng)讓出（yield）控制流，并記住自身當(dāng)前的狀態(tài)，以便在控制流返回時(shí)能從上次讓出的位置恢復(fù)（resume）執(zhí)行。

簡(jiǎn)而言之，協(xié)程的核心思想就在于執(zhí)行者對(duì)控制流的 “主動(dòng)讓出” 和 “恢復(fù)”。相對(duì)于，線程此類的 “搶占式調(diào)度” 而言，協(xié)程是一種 “協(xié)作式調(diào)度” 方式。

在這里插入圖片描述

協(xié)程的應(yīng)用場(chǎng)景

搶占式調(diào)度的缺點(diǎn)

在 I/O 密集型場(chǎng)景中，搶占式調(diào)度的解決方案是 “異步 + 回調(diào)” 機(jī)制。

在這里插入圖片描述

其存在的問(wèn)題是，在某些場(chǎng)景中會(huì)使得整個(gè)程序的可讀性非常差。以圖片下載為例，圖片服務(wù)中臺(tái)提供了異步接口，發(fā)起者請(qǐng)求之后立即返回，圖片服務(wù)此時(shí)給了發(fā)起者一個(gè)唯一標(biāo)識(shí) ID，等圖片服務(wù)完成下載后把結(jié)果放到一個(gè)消息隊(duì)列，此時(shí)需要發(fā)起者不斷消費(fèi)這個(gè) MQ 才能拿到下載是否完成的結(jié)果。

在這里插入圖片描述

可見(jiàn)，整體的邏輯被拆分為了好幾個(gè)部分，各個(gè)子部分都會(huì)存在狀態(tài)的遷移，日后必然是 BUG 的高發(fā)地。

在這里插入圖片描述

用戶態(tài)協(xié)同調(diào)度的優(yōu)勢(shì)

而隨著網(wǎng)絡(luò)技術(shù)的發(fā)展和高并發(fā)要求，協(xié)程所能夠提供的用戶態(tài)協(xié)同調(diào)度機(jī)制的優(yōu)勢(shì)，在網(wǎng)絡(luò)操作、文件操作、數(shù)據(jù)庫(kù)操作、消息隊(duì)列操作等重 I/O 操作場(chǎng)景中逐漸被挖掘。

在這里插入圖片描述

協(xié)程將 I/O 的處理權(quán)從內(nèi)核態(tài)的操作系統(tǒng)交還給用戶態(tài)的程序自身。用戶態(tài)程序在執(zhí)行 I/O 時(shí)，主動(dòng)的通過(guò) yield（讓出）CPU 的執(zhí)行權(quán)給其他協(xié)程，多個(gè)協(xié)程之間處于平等、對(duì)稱、合作的關(guān)系。

協(xié)程的運(yùn)行原理

當(dāng)程序運(yùn)行時(shí)，操作系統(tǒng)會(huì)為每個(gè)程序分配一塊同等大小的虛擬內(nèi)存空間，并將程序的代碼和所有靜態(tài)數(shù)據(jù)加載到其中。然后，創(chuàng)建和初始化 Stack 存儲(chǔ)，用于儲(chǔ)存程序的局部變量，函數(shù)參數(shù)和返回地址；創(chuàng)建和初始化 Heap 內(nèi)存；創(chuàng)建和初始化 I/O 相關(guān)的任務(wù)。當(dāng)前期準(zhǔn)備工作完成后，操作系統(tǒng)將 CPU 的控制權(quán)移交給新創(chuàng)建的進(jìn)程，進(jìn)程開(kāi)始運(yùn)行。

在這里插入圖片描述

一個(gè)進(jìn)程可以有一個(gè)或多個(gè)線程，同一進(jìn)程中的多個(gè)線程將共享該進(jìn)程中的全部系統(tǒng)資源，如：虛擬地址空間，文件描述符和信號(hào)處理等等。但同一進(jìn)程中的多個(gè)線程有各自的調(diào)用棧和線程本地存儲(chǔ)。

在這里插入圖片描述

協(xié)程是一種比線程更加輕量級(jí)的存在，協(xié)程不是被操作系統(tǒng)內(nèi)核所管理，而完全是由用戶態(tài)程序所控制。協(xié)程與線程以及進(jìn)程的關(guān)系如下圖所示?？梢?jiàn)，協(xié)程自身無(wú)法利用多核，需要配合進(jìn)程來(lái)使用才可以在多核平臺(tái)上發(fā)揮作用。

在這里插入圖片描述

協(xié)程之間的切換不需要涉及任何 System Call（系統(tǒng)調(diào)用）或任何阻塞調(diào)用。
協(xié)程只在一個(gè)線程中執(zhí)行，切換由用戶態(tài)控制，而線程的阻塞狀態(tài)是由操作系統(tǒng)內(nèi)核來(lái)完成的，因此協(xié)程相比線程節(jié)省線程創(chuàng)建和切換的開(kāi)銷。
協(xié)程中不存在同時(shí)寫變量的沖突，因此，也就不需要用來(lái)守衛(wèi)關(guān)鍵區(qū)塊的同步性原語(yǔ)，比如：互斥鎖、信號(hào)量等，并且不需要來(lái)自操作系統(tǒng)的支持。

協(xié)程通過(guò) “掛起點(diǎn)” 來(lái)主動(dòng) yield（讓出）CPU，并保存自身的狀態(tài)，等候恢復(fù)。例如：首先在 funcA 函數(shù)中執(zhí)行，運(yùn)行一段時(shí)間后調(diào)用協(xié)程，協(xié)程開(kāi)始執(zhí)行，直到第一個(gè)掛起點(diǎn)，此后就像普通函數(shù)一樣返回 funcA 函數(shù)。 funcA 函數(shù)執(zhí)行一些代碼后再次調(diào)用該協(xié)程，注意，協(xié)程這時(shí)就和普通函數(shù)不一樣了。協(xié)程并不是從第一條指令開(kāi)始執(zhí)行而是從上一次的掛起點(diǎn)開(kāi)始執(zhí)行，執(zhí)行一段時(shí)間后遇到第二個(gè)掛起點(diǎn)，這時(shí)協(xié)程再次像普通函數(shù)一樣返回 funcA 函數(shù)，funcA 函數(shù)執(zhí)行一段時(shí)間后整個(gè)程序結(jié)束。

在這里插入圖片描述

可見(jiàn)，協(xié)程之所可以能夠 “主動(dòng)讓出” 和 “被恢復(fù)”，是解析器在函數(shù)運(yùn)行時(shí)堆棧中保存了其運(yùn)行的 Context（上下文）。

在這里插入圖片描述

Python 中的協(xié)程

Python 對(duì)協(xié)程的支持經(jīng)歷了多個(gè)版本：

Python2.x 對(duì)協(xié)程的支持比較有限，通過(guò) yield 關(guān)鍵字支持的生成器實(shí)現(xiàn)了一部分協(xié)程的功能但不完全。
第三方庫(kù) gevent 對(duì)協(xié)程有更好的支持。
Python3.4 中提供了 asyncio 模塊。
Python3.5 中引入了 async/await 關(guān)鍵字。
Python3.6 中 asyncio 模塊更加完善和穩(wěn)定。
Python3.7 中內(nèi)置了 async/await 關(guān)鍵字。

async/await 的示例程序：

import asyncio
from pathlib import Path
import logging
from urllib.request import urlopen, Request
import os
from time import time
import aiohttp
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
CODEFLEX_IMAGES_URLS = ['https://codeflex.co/wp-content/uploads/2021/01/pandas-dataframe-python-1024x512.png',
                        'https://codeflex.co/wp-content/uploads/2021/02/github-actions-deployment-to-eks-with-kustomize-1024x536.jpg',
                        'https://codeflex.co/wp-content/uploads/2021/02/boto3-s3-multipart-upload-1024x536.jpg',
                        'https://codeflex.co/wp-content/uploads/2018/02/kafka-cluster-architecture.jpg',
                        'https://codeflex.co/wp-content/uploads/2016/09/redis-cluster-topology.png']
async def download_image_async(session, dir, img_url):
    download_path = dir / os.path.basename(img_url)
    async with session.get(img_url) as response:
        with download_path.open('wb') as f:
            while True:
                # 在 async 函數(shù)中使用 await 關(guān)鍵字表示等待 task 執(zhí)行完成，也就是等待 yeild 讓出控制權(quán)。
                # 同時(shí)，asyncio 使用事件循環(huán) event_loop 來(lái)實(shí)現(xiàn)整個(gè)過(guò)程。
                chunk = await response.content.read(512)
                if not chunk:
                    break
                f.write(chunk)
    logger.info('Downloaded: ' + img_url)
# 使用 async 關(guān)鍵字聲明一個(gè)異步/協(xié)程函數(shù)。
# 調(diào)用該函數(shù)時(shí)，并不會(huì)立即運(yùn)行，而是返回一個(gè)協(xié)程對(duì)象，后續(xù)在 event_loop 中執(zhí)行。
async def main():
    images_dir = Path("codeflex_images")
    Path("codeflex_images").mkdir(parents=False, exist_ok=True)
    async with aiohttp.ClientSession() as session:
        tasks = [(download_image_async(session, images_dir, img_url)) for img_url in CODEFLEX_IMAGES_URLS]
        await asyncio.gather(*tasks, return_exceptions=True)
if __name__ == '__main__':
    start = time()
    # event_loop 事件循環(huán)充當(dāng)管理者的角色，將控制權(quán)在幾個(gè)協(xié)程函數(shù)之間切換。
    event_loop = asyncio.get_event_loop()
    try:
        event_loop.run_until_complete(main())
    finally:
        event_loop.close()
    logger.info('Download time: %s seconds', time() - start)