快捷導(dǎo)航

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

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

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

協(xié)程

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

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

在這里插入圖片描述

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

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

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

在這里插入圖片描述

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

在這里插入圖片描述

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

在這里插入圖片描述

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

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

在這里插入圖片描述

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

協(xié)程的運行原理

當(dāng)程序運行時，操作系統(tǒng)會為每個程序分配一塊同等大小的虛擬內(nèi)存空間，并將程序的代碼和所有靜態(tài)數(shù)據(jù)加載到其中。然后，創(chuàng)建和初始化 Stack 存儲，用于儲存程序的局部變量，函數(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)建的進程，進程開始運行。

在這里插入圖片描述

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

在這里插入圖片描述

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

在這里插入圖片描述

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

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

在這里插入圖片描述

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

在這里插入圖片描述

Python 中的協(xié)程

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

Python2.x 對協(xié)程的支持比較有限，通過 yield 關(guān)鍵字支持的生成器實現(xiàn)了一部分協(xié)程的功能但不完全。
第三方庫 gevent 對協(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)。
                # 同時，asyncio 使用事件循環(huán) event_loop 來實現(xiàn)整個過程。
                chunk = await response.content.read(512)
                if not chunk:
                    break
                f.write(chunk)
    logger.info('Downloaded: ' + img_url)
# 使用 async 關(guān)鍵字聲明一個異步/協(xié)程函數(shù)。
# 調(diào)用該函數(shù)時，并不會立即運行，而是返回一個協(xié)程對象，后續(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)在幾個協(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)