python進階（25）協程

2022 年 2 月 22 日
筆記
Python進階

協程的定義

協程(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 = [
        '//www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        '//www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        '//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 = ["//www.baidu.com", "//www.taobao.com", "//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())

Tags: Python進階

python進階（25）協程

協程的定義

協程和執行緒差異

協程的標準

協程的優點

協程的缺點

python中實現協程的方式

async&await關鍵字

事件循環

協程函數和協程對象

await

Task對象

asyncio.Future對象

futures.Future對象

非同步迭代器

什麼是非同步迭代器？

什麼是非同步可迭代對象？

非同步上下文管理器

uvloop

非同步redis

非同步MySQL

爬蟲

VirMach 便宜 VPS

QNews

python進階（25）協程

協程的定義

協程和執行緒差異

協程的標準

協程的優點

協程的缺點

python中實現協程的方式

async&await關鍵字

事件循環

協程函數和協程對象

await

Task對象

asyncio.Future對象

futures.Future對象

非同步迭代器

什麼是非同步迭代器？

什麼是非同步可迭代對象？

非同步上下文管理器

uvloop

非同步redis

非同步MySQL

爬蟲

分享此文：

Related Posts

kill 指令的執行原理

uniapp 滑動切換

華碩無畏Pro 15輕薄筆電秒殺：2.8K OLED螢幕只要4999

【Python自動化Excel】pandas處理Excel的「分分合合」

VirMach 便宜 VPS

QNews

熱門搜尋