快捷導(dǎo)航

詳解python之異步編程

更新時(shí)間：2021年12月24日 08:54:03 作者：Python_xiaowu

這篇文章主要為大家介紹了python之異步編程，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來(lái)幫助<BR>

一、異步編程概述

異步編程是一種并發(fā)編程的模式，其關(guān)注點(diǎn)是通過(guò)調(diào)度不同任務(wù)之間的執(zhí)行和等待時(shí)間，通過(guò)減少處理器的閑置時(shí)間來(lái)達(dá)到減少整個(gè)程序的執(zhí)行時(shí)間；異步編程跟同步編程模型最大的不同就是其任務(wù)的切換，當(dāng)遇到一個(gè)需要等待長(zhǎng)時(shí)間執(zhí)行的任務(wù)的時(shí)候，我們可以切換到其他的任務(wù)執(zhí)行；

與多線(xiàn)程和多進(jìn)程編程模型相比，異步編程只是在同一個(gè)線(xiàn)程之內(nèi)的的任務(wù)調(diào)度，無(wú)法充分利用多核CPU的優(yōu)勢(shì)，所以特別適合IO阻塞性任務(wù)；

python 版本 3.9.5

二、python的異步框架模型

python提供了asyncio模塊來(lái)支持異步編程，其中涉及到coroutines、event loops、futures三個(gè)重要概念;

event loops主要負(fù)責(zé)跟蹤和調(diào)度所有異步任務(wù)，編排具體的某個(gè)時(shí)間點(diǎn)執(zhí)行的任務(wù);

coroutines是對(duì)具體執(zhí)行任務(wù)的封裝，是一個(gè)可以在執(zhí)行中暫停并切換到event loops執(zhí)行流程的特殊類(lèi)型的函數(shù);其一般還需要?jiǎng)?chuàng)建task才能被event loops調(diào)度;

futures負(fù)責(zé)承載coroutines的執(zhí)行結(jié)果，其隨著任務(wù)在event loops中的初始化而創(chuàng)建，并隨著任務(wù)的執(zhí)行來(lái)記錄任務(wù)的執(zhí)行狀態(tài);

異步編程框架的整個(gè)執(zhí)行過(guò)程涉及三者的緊密協(xié)作；

首先事件循環(huán)啟動(dòng)之后，會(huì)從任務(wù)隊(duì)列獲取第一個(gè)要執(zhí)行的coroutine，并隨之創(chuàng)建對(duì)應(yīng)task和future;

然后隨著task的執(zhí)行，當(dāng)遇到coroutine內(nèi)部需要切換任務(wù)的地方，task的執(zhí)行就會(huì)暫停并釋放執(zhí)行線(xiàn)程給event loop，event loop接著會(huì)獲取下一個(gè)待執(zhí)行的coroutine，并進(jìn)行相關(guān)的初始化之后，執(zhí)行這個(gè)task;

隨著event loop執(zhí)行完隊(duì)列中的最后一個(gè)coroutine才會(huì)切換到第一個(gè)coroutine;

隨著task的執(zhí)行結(jié)束，event loops會(huì)將task清除出隊(duì)列，對(duì)應(yīng)的執(zhí)行結(jié)果會(huì)同步到future中，這個(gè)過(guò)程會(huì)持續(xù)到所有的task執(zhí)行結(jié)束;

三、順序執(zhí)行多個(gè)可重疊的任務(wù)

每個(gè)任務(wù)執(zhí)行中間會(huì)暫停給定的時(shí)間，循序執(zhí)行的時(shí)間就是每個(gè)任務(wù)執(zhí)行的時(shí)間加和；

import time
def count_down(name, delay):
    indents = (ord(name) - ord('A')) * '\t'
    n = 3
    while n:
        time.sleep(delay)
        duration = time.perf_counter() - start
        print('-' * 40)
        print(f'{duration:.4f} \t{indents}{name} = {n}')
        n -= 1
start = time.perf_counter()
count_down('A', 1)
count_down('B', 0.8)
count_down('C', 0.5)
print('-' * 40)
print('Done')
# ----------------------------------------
# 1.0010 	A = 3
# ----------------------------------------
# 2.0019 	A = 2
# ----------------------------------------
# 3.0030 	A = 1
# ----------------------------------------
# 3.8040 		B = 3
# ----------------------------------------
# 4.6050 		B = 2
# ----------------------------------------
# 5.4059 		B = 1
# ----------------------------------------
# 5.9065 			C = 3
# ----------------------------------------
# 6.4072 			C = 2
# ----------------------------------------
# 6.9078 			C = 1
# ----------------------------------------
# Done

四、異步化同步代碼

python在語(yǔ)法上提供了async、await兩個(gè)關(guān)鍵字來(lái)簡(jiǎn)化將同步代碼修改為異步;

async使用在函數(shù)的def關(guān)鍵字前邊，標(biāo)記這是一個(gè)coroutine函數(shù);

await用在conroutine里邊，用于標(biāo)記需要暫停釋放執(zhí)行流程給event loops;

await 后邊的表達(dá)式需要返回waitable的對(duì)象，例如conroutine、task、future等;

asyncio模塊主要提供了操作event loop的方式;

我們可以通過(guò)async將count_down標(biāo)記為coroutine，然后使用await和asyncio.sleep來(lái)實(shí)現(xiàn)異步的暫停，從而將控制權(quán)交給event loop；

async def count_down(name, delay, start):
    indents = (ord(name) - ord('A')) * '\t'
    n = 3
    while n:
        await asyncio.sleep(delay)
        duration = time.perf_counter() - start
        print('-' * 40)
        print(f'{duration:.4f} \t{indents}{name} = {n}')
        n -= 1

我們定義一個(gè)異步的main方法，主要完成task的創(chuàng)建和等待任務(wù)執(zhí)行結(jié)束；

async def main():
    start = time.perf_counter()
    tasks = [asyncio.create_task(count_down(name,delay,start)) for name, delay in [('A', 1),('B', 0.8),('C', 0.5)]]
    await asyncio.wait(tasks)
    print('-' * 40)
    print('Done')

執(zhí)行我們可以看到時(shí)間已經(jīng)變?yōu)榱藞?zhí)行時(shí)間最長(zhǎng)的任務(wù)的時(shí)間了；

asyncio.run(main())
# ----------------------------------------
# 0.5010 			C = 3
# ----------------------------------------
# 0.8016 		B = 3
# ----------------------------------------
# 1.0011 	A = 3
# ----------------------------------------
# 1.0013 			C = 2
# ----------------------------------------
# 1.5021 			C = 1
# ----------------------------------------
# 1.6026 		B = 2
# ----------------------------------------
# 2.0025 	A = 2
# ----------------------------------------
# 2.4042 		B = 1
# ----------------------------------------
# 3.0038 	A = 1
# ----------------------------------------
# Done

五、使用多線(xiàn)程克服具體任務(wù)的異步限制

異步編程要求具體的任務(wù)必須是coroutine，也就是要求方法是異步的，否則只有任務(wù)執(zhí)行完了，才能將控制權(quán)釋放給event loop；

python中的concurent.futures提供了ThreadPoolExecutor和ProcessPoolExecutor，可以直接在異步編程中使用，從而可以在單獨(dú)的線(xiàn)程或者進(jìn)程至今任務(wù)；

import time
import asyncio
from concurrent.futures import ThreadPoolExecutor
def count_down(name, delay, start):
    indents = (ord(name) - ord('A')) * '\t'
    n = 3
    while n:
        time.sleep(delay)
        duration = time.perf_counter() - start
        print('-'*40)
        print(f'{duration:.4f} \t{indents}{name} = {n}')
        n -=1
async def main():
    start = time.perf_counter()
    loop = asyncio.get_running_loop()
    executor = ThreadPoolExecutor(max_workers=3)
    fs = [
       loop.run_in_executor(executor, count_down, *args)  for args in [('A', 1, start), ('B', 0.8, start), ('C', 0.5, start)]
    ]
    await asyncio.wait(fs)
    print('-'*40)
    print('Done.')
asyncio.run(main())
# ----------------------------------------
# 0.5087 			C = 3
# ----------------------------------------
# 0.8196 		B = 3
# ----------------------------------------
# 1.0073 	A = 3
# ----------------------------------------
# 1.0234 			C = 2
# ----------------------------------------
# 1.5350 			C = 1
# ----------------------------------------
# 1.6303 		B = 2
# ----------------------------------------
# 2.0193 	A = 2
# ----------------------------------------
# 2.4406 		B = 1
# ----------------------------------------
# 3.0210 	A = 1
# ----------------------------------------
# Done.