python進階之協程你瞭解嗎

2022-02-22 19:00:26

協程的定義

協程(Coroutine)，又稱微執行緒，纖程。(協程是一種使用者態的輕量級執行緒)

作用：在執行 A 函數的時候，可以隨時中斷，去執行 B 函數，然後中斷B函數，繼續執行 A 函數（可以自動切換），但這一過程並不是函數呼叫（沒有呼叫語句），過程很像多執行緒，然而協程只有一個執行緒在執行

通俗的理解：在一個執行緒中的某個函數，可以在任何地方儲存當前函數的一些臨時變數等資訊，然後切換到另外一個函數中執行，注意不是通過呼叫函數的方式做到的，並且切換的次數以及什麼時候再切換到原來的函數都由開發者自己確定

協程和執行緒差異

在實現多工時, 執行緒切換從系統層面遠不止儲存和恢復 CPU上下文這麼簡單。作業系統為了程式執行的高效性每個執行緒都有自己快取Cache等等資料，作業系統還會幫你做這些資料的恢復操作。所以執行緒的切換非常耗效能。但是協程的切換隻是單純的操作CPU的上下文，所以一秒鐘切換個上百萬次系統都抗的住。

協程的標準

必須在只有一個單執行緒裡實現並行

修改共用資料不需加鎖
使用者程式裡自己儲存多個控制流的上下文棧
一個協程遇到 IO 操作自動切換到其它協程

協程的優點

由於自身帶有上下文和棧，無需執行緒上下文切換的開銷,屬於程式級別的切換，作業系統完全感知不到，因而更加輕量級
無需原子操作的鎖定及同步的開銷
方便切換控制流，簡化程式設計模型
單執行緒內就可以實現並行的效果，最大限度地利用 cpu，且可延伸性高，成本低

協程的缺點

無法利用多核資源：協程的本質是個單執行緒,它不能同時將單個 CPU 的多個核用上,協程需要和程序配合才能執行在多 CPU 上
進行阻塞（Blocking）操作（如 IO 時）會阻塞掉整個程式
計算型的操作，利用協程來回切換執行，沒有任何意義，來回切換並儲存狀態反倒會降低效能。

python中實現協程的方式

greenlet，是一個第三方模組，用於實現協程程式碼（Gevent協程就是基於greenlet實現）
yield，生成器，藉助生成器的特點也可以實現協程程式碼。
asyncio，在Python3.4中引入的模組用於編寫協程程式碼。
async & awiat，在Python3.5中引入的兩個關鍵字，結合asyncio模組可以更方便的編寫協程程式碼（推薦）。

async&await關鍵字

協程實現的方式有多種，目前最流行的方式就是async&await，其他的方式瞭解即可，本文就介紹最流行的一種方式

使用協程需要了解2個，事件迴圈和定義協程函數

事件迴圈

事件迴圈是一種處理多並行量的有效方式，可以理解為死迴圈，迴圈過程中去檢測並執行某些程式碼，我們來看下面的虛擬碼

任務列表 = [任務1， 任務2， 任務3............]
while True:
    可執行的任務列表，已完成的任務列表 = 去任務列表中檢查所有的任務，將'可執行'和'已完成'的任務返回
    for 就緒任務 in 可執行的任務列表：
        執行已就緒的任務
    for 已完成的任務 in 已完成的任務列表：
        在任務列表中移除 已完成的任務
    如果任務列表中的任務都已完成，則終止迴圈

上面的虛擬碼的意思：獲取事件迴圈中，然後不斷監聽任務列表，有任務就執行，執行完成的任務就移除，直到任務列表中的所有任務都完成，終止迴圈

使用事件迴圈的好處：使得程式設計師不用控制任務的新增、刪除和事件的控制

程式碼中的寫法如下：

import asyncio
# 獲取事件迴圈
loop = asyncio.get_event_loop()
# 將任務放到`任務列表`，監聽事件迴圈
loop.run_until_complete(任務)
# 關閉事件
loop.close()

協程函數和協程物件

想要定義協程函數，格式：async def 函數名協程物件：執行協程函數()得到的協程物件

# 定義協程函數
async def func():
    pass
# 建立協程物件
result = func()

注意：執行協程函數，建立協程物件，函數程式碼是不會執行的，如果想要執行協程函數的內部程式碼，必須要將協程物件交給事件迴圈來處理，看如下程式碼

import asyncio
async def func():
    print("你好")
result = func()
# 方式1
loop = asyncio.get_event_loop()
loop.run_until_complete(result)
# 方式2
asyncio.run(result)  # python3.7寫法

await

await是一個只能在協程函數中使用的關鍵字，用於遇到IO操作時掛起當前協程（任務），當前協程（任務）掛起過程中事件迴圈可以去執行其他的協程（任務），當前協程IO處理完成時，可以再次切換回來執行await之後的程式碼。

舉例：我們建立了2個任務，一個下載圖片，一個下載視訊，我們先執行下載圖片任務，這時遇到io操作，正常情況下會等待圖片下載完，但await可以先掛起下載圖片任務，然後自動切換到下載視訊任務

使用方法：await + 可等待物件（協程物件、Future物件、Task物件）

案例1

import asyncio
async def func():
    print("執行協程函數內部程式碼")
    # 遇到IO操作掛起當前協程（任務），等IO操作完成之後再繼續往下執行。
    # 當前協程掛起時，事件迴圈可以去執行其他協程（任務）。
    response = await asyncio.sleep(2)
    print("IO請求結束，結果為：", response)
result = func()
asyncio.run(result)

案列2

import asyncio
async def others():
    print("start")  # ④列印start
    await asyncio.sleep(2)  # ⑤等待耗時2秒，這過程中可以切換到其他協程中去
    print("end")  # ⑥列印end
    return '返回值'
async def func():
    print("執行協程函數內部程式碼")  # ②執行協程函數，列印print程式碼
    response = await others()  # ③等待協程函數others
    print(f"io請求結束，結果為{response}")  # ⑦等待others結束後列印print語句
if __name__ == '__main__':
    asyncio.run(func())  # ①協程函數放入事件迴圈中執行

上述的所有範例都只是建立了一個任務，即：事件迴圈的任務列表中只有一個任務，所以在IO等待時無法演示切換到其他任務效果。在程式想要建立多個任務物件，需要使用Task物件來實現。

Task物件

Tasks用於並行排程協程，通過asyncio.create_task(協程物件)的方式建立Task物件，這樣可以讓協程加入事件迴圈中等待被排程執行。除了使用 asyncio.create_task() 函數以外，還可以用低層級的loop.create_task()或 ensure_future() 函數。不建議手動範例化 Task 物件。

本質上是將協程物件封裝成task物件，並將協程立即加入事件迴圈，同時追蹤協程的狀態。

注意：asyncio.create_task() 函數在 Python 3.7 中被加入。在 Python 3.7 之前，可以改用低層級的 asyncio.ensure_future() 函數。

案例1

import asyncio
async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"
async def main():
    print("main開始")
    # 建立協程，將協程封裝到一個Task物件中並立即新增到事件迴圈的任務列表中，等待事件迴圈去執行（預設是就緒狀態）。
    task1 = asyncio.create_task(func())
    # 建立協程，將協程封裝到一個Task物件中並立即新增到事件迴圈的任務列表中，等待事件迴圈去執行（預設是就緒狀態）。
    task2 = asyncio.create_task(func())
    print("main結束")
    # 當執行某協程遇到IO操作時，會自動化切換執行其他任務。
    # 此處的await是等待相對應的協程全都執行完畢並獲取結果
    ret1 = await task1
    ret2 = await task2
    print(ret1, ret2)
asyncio.run(main())

案例await+任務列表（用的最多）

import asyncio
async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"
async def main():
    print("main開始")
    # 建立協程，將協程封裝到Task物件中並新增到事件迴圈的任務列表中，等待事件迴圈去執行（預設是就緒狀態）。
    # 在呼叫
    task_list = [asyncio.create_task(func()), asyncio.create_task(func())]
    print("main結束")
    # 當執行某協程遇到IO操作時，會自動化切換執行其他任務。
    # 此處的await是等待所有協程執行完畢，並將所有協程的返回值儲存到done
    # 如果設定了timeout值，則意味著此處最多等待的秒，完成的協程返回值寫入到done中，未完成則寫到pending中。
    done, pending = await asyncio.wait(task_list)
    print(done)
asyncio.run(main())

注意： asyncio.wait 原始碼內部會對列表中的每個協程執行ensure_future從而封裝為Task物件，所以在和wait配合使用時task_list的值為[func(),func()] 也是可以的。

asyncio.Future物件

asyncio中的Future物件是一個相對更偏向底層的可物件，通常我們不會直接用到這個物件，而是直接使用Task物件來完成任務的並和狀態的追蹤。（ Task 是 Futrue的子類）

Future為我們提供了非同步程式設計中的最終結果的處理（Task類也具備狀態處理的功能）

案例1

async def main():
    # 獲取當前事件迴圈
    loop = asyncio.get_running_loop()
    # 建立一個任務（Future物件），這個任務什麼都不幹。
    fut = loop.create_future()
    # 等待任務最終結果（Future物件），沒有結果則會一直等下去。
    await fut
asyncio.run(main())

結果就是程式一直在等待，無法結束

案例2

import asyncio
async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result("666")
async def main():
    # 獲取當前事件迴圈
    loop = asyncio.get_running_loop()
    # 建立一個任務（Future物件），沒繫結任何行為，則這個任務永遠不知道什麼時候結束。
    fut = loop.create_future()
    # 建立一個任務（Task物件），繫結了set_after函數，函數內部在2s之後，會給fut賦值。
    # 即手動設定future任務的最終結果，那麼fut就可以結束了。
    await loop.create_task(set_after(fut))
    # 等待 Future物件獲取 最終結果，否則一直等下去
    data = await fut
    print(data)
asyncio.run(main())

Future物件本身函數進行繫結，所以想要讓事件迴圈獲取Future的結果，則需要手動設定。而Task物件繼承了Future物件，其實就對Future進行擴充套件，他可以實現在對應繫結的函數執行完成之後，自動執行set_result，從而實現自動結束。

雖然，平時使用的是Task物件，但對於結果的處理本質是基於Future物件來實現的。

futures.Future物件

在Python的concurrent.futures模組中也有一個Future物件，這個物件是基於執行緒池和程序池實現非同步操作時使用的物件。

import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor
def func(value):
    time.sleep(1)
    print(value)
pool = ThreadPoolExecutor(max_workers=5)
# 或 pool = ProcessPoolExecutor(max_workers=5)
for i in range(10):
    fut = pool.submit(func, i)
    print(fut)

兩個Future物件是不同的，他們是為不同的應用場景而設計，例如：concurrent.futures.Future不支援await語法等。

在Python提供了一個將futures.Future 物件包裝成asyncio.Future物件的函數 asynic.wrap_future。

接下里你肯定問：為什麼python會提供這種功能？

其實，一般在程式開發中我們要麼統一使用 asycio 的協程實現非同步操作、要麼都使用程序池和執行緒池實現非同步操作。但如果協程的非同步和程序池/執行緒池的非同步混搭時，那麼就會用到此功能了。

import time
import asyncio
import concurrent.futures
def func1():
    # 某個耗時操作
    time.sleep(2)
    return "OK"
async def main():
    loop = asyncio.get_running_loop()
    # 方式1. Run in the default loop's executor ( 預設ThreadPoolExecutor )
    # 第一步：內部會先呼叫 ThreadPoolExecutor 的 submit 方法去執行緒池中申請一個執行緒去執行func1函數，並返回一個concurrent.futures.Future物件
    # 第二步：呼叫asyncio.wrap_future將concurrent.futures.Future物件包裝為asycio.Future物件。
    # 因為concurrent.futures.Future物件不支援await語法，所以需要包裝為 asycio.Future物件 才能使用。
    fut = loop.run_in_executor(None, func1)
    result = await fut
    print('default thread pool', result)
    # 方式2. Run in a custom thread pool:
    # with concurrent.futures.ThreadPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom thread pool', result)
    # 方式3. Run in a custom process pool:
    # with concurrent.futures.ProcessPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom process pool', result)
asyncio.run(main())

應用場景：當專案以協程式的非同步程式設計開發時，如果要使用一個第三方模組，而第三方模組不支援協程方式非同步程式設計時，就需要用到這個功能，例如requests模組：

import asyncio
import requests
async def download_image(url):
    # 傳送網路請求，下載圖片（遇到網路下載圖片的IO請求，自動化切換到其他任務）
    print("開始下載:", url)
    loop = asyncio.get_event_loop()
    # requests模組預設不支援非同步操作，所以就使用執行緒池來配合實現了。
    future = loop.run_in_executor(None, requests.get, url)
    response = await future
    print('下載完成')
    # 圖片儲存到本地檔案
    file_name = url.rsplit('_')[-1]
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)
if __name__ == '__main__':
    url_list = [
        'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
    ]
    tasks = [download_image(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete( asyncio.wait(tasks) )

非同步迭代器

什麼是非同步迭代器？

實現了__aiter__()和__anext__()方法的物件。__anext__ 必須返回一個 awaitable 物件。async for 會處理非同步迭代器的__anext__()方法所返回的可等待物件，直到其引發一個 StopAsyncIteration 異常。

什麼是非同步可迭代物件？

可在 async for 語句中被使用的物件。必須通過它的 __aiter__() 方法返回一個 asynchronous iterator。

import asyncio
class Reader(object):
    """ 自定義非同步迭代器（同時也是非同步可迭代物件） """
    def __init__(self):
        self.count = 0
    async def readline(self):
        # await asyncio.sleep(1)
        self.count += 1
        if self.count == 100:
            return None
        return self.count
    def __aiter__(self):
        return self
    async def __anext__(self):
        val = await self.readline()
        if val == None:
            raise StopAsyncIteration
        return val
async def func():
    # 建立非同步可迭代物件
    async_iter = Reader()
    # async for 必須要放在async def函數內，否則語法錯誤。
    async for item in async_iter:
        print(item)
asyncio.run(func())

非同步迭代器其實沒什麼太大的作用，只是支援了async for語法而已。

非同步上下文管理器

此種物件通過定義 __aenter__() 和 __aexit__() 方法來對 async with 語句中的環境進行控制

import asyncio
class AsyncContextManager:
    def __init__(self):
        self.conn = None
    async def do_something(self):
        # 非同步運算元據庫
        return 666
    async def __aenter__(self):
        # 非同步連結資料庫
        self.conn = await asyncio.sleep(1)
        return self
    async def __aexit__(self, exc_type, exc, tb):
        # 非同步關閉資料庫連結
        await asyncio.sleep(1)
async def func():
    async with AsyncContextManager() as f:
        result = await f.do_something()
        print(result)
asyncio.run(func())

這個非同步的上下文管理器還是比較有用的，平時在開發過程中開啟、處理、關閉操作時，就可以用這種方式來處理。

uvloop

uvloop是 asyncio 中的事件迴圈的替代方案，替換後可以使得asyncio效能提高。事實上，uvloop要比nodejs、gevent等其他python非同步框架至少要快2倍，效能可以比肩Go語言。

安裝uvloop

pip3 install uvloop

在專案中想要使用uvloop替換asyncio的事件迴圈也非常簡單，只要在程式碼中這麼做就行。

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 編寫asyncio的程式碼，與之前寫的程式碼一致。
# 內部的事件迴圈自動化會變為uvloop
asyncio.run(...)

注意：知名的asgi uvicorn內部就是使用的uvloop的事件迴圈。

非同步redis

當通過python去操作redis時，連結、設定值、獲取值這些都涉及網路IO請求，使用asycio非同步的方式可以在IO等待時去做一些其他任務，從而提升效能。

安裝Python非同步操作redis模組

pip3 install aioredis

案例:連線多個redis做操作（遇到IO會切換其他任務，提供了效能）

import asyncio
import aioredis
async def execute(address, password):
    print("開始執行", address)
    # 網路IO操作：先去連線 77.95.4.197:6379，遇到IO則自動切換任務，去連線77.95.4.198:6379
    redis = await aioredis.create_redis_pool(address, password=password)
    # 網路IO操作：遇到IO會自動切換任務
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)
    # 網路IO操作：遇到IO會自動切換任務
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)
    redis.close()
    # 網路IO操作：遇到IO會自動切換任務
    await redis.wait_closed()
    print("結束", address)
task_list = [
    execute('redis://77.95.4.197:6379', "123456"),
    execute('redis://77.95.4.198:6379', "123456")
]
asyncio.run(asyncio.wait(task_list))

非同步MySQL

當通過python去操作MySQL時，連線、執行SQL、關閉都涉及網路IO請求，使用asycio非同步的方式可以在IO等待時去做一些其他任務，從而提升效能。

安裝Python非同步操作redis模組

pip3 install aiomysql

案例

import asyncio
import aiomysql
async def execute(host, password):
    print("開始", host)
    # 網路IO操作：先去連線 77.95.40.197，遇到IO則自動切換任務，去連線77.95.40.198:6379
    conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')
    # 網路IO操作：遇到IO會自動切換任務
    cur = await conn.cursor()
    # 網路IO操作：遇到IO會自動切換任務
    await cur.execute("SELECT Host,User FROM user")
    # 網路IO操作：遇到IO會自動切換任務
    result = await cur.fetchall()
    print(result)
    # 網路IO操作：遇到IO會自動切換任務
    await cur.close()
    conn.close()
    print("結束", host)
task_list = [
    execute('77.95.40.197', "123456"),
    execute('77.95.40.198', "123456")
]
asyncio.run(asyncio.wait(task_list))

爬蟲

在編寫爬蟲應用時，需要通過網路IO去請求目標資料，這種情況適合使用非同步程式設計來提升效能，接下來我們使用支援非同步程式設計的aiohttp模組來實現。

安裝aiohttp模組

pip3 install aiohttp

案例

import aiohttp
import asyncio
async def fetch(session, url):
    print(f"傳送請求：{url}")
    async with session.get(url, verify_ssl=False) as response:
        text = await response.text()
        print("得到結果：", url, len(text))
async def main():
    async with aiohttp.ClientSession() as session:
        url_list = ["http://www.baidu.com", "http://www.taobao.com", "http://www.jd.com"]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)
if __name__ == '__main__':
    asyncio.run(main())