當前位置：首頁 > 编程语言 > python >内容正文

python

python异步编程视频_asyncio异步编程【含视频教程】

發(fā)布時間：2024/10/8 python 41 豆豆

生活随笔收集整理的這篇文章主要介紹了 python异步编程视频_asyncio异步编程【含视频教程】小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

Python

Python開發(fā)

Python語言

asyncio異步編程【含視頻教程】

不知道你是否發(fā)現(xiàn)，身邊聊異步的人越來越多了，比如：FastAPI、Tornado、Sanic、Django 3、aiohttp等。

聽說異步如何如何牛逼？性能如何吊炸天。。。。但他到底是咋回事呢？

本節(jié)要跟大家一起聊聊關(guān)于asyncio異步的那些事！

1.協(xié)程

想學asyncio，得先了解協(xié)程，協(xié)程是根本呀！

協(xié)程(Coroutine)，也可以被稱為微線程，是一種用戶態(tài)內(nèi)的上下文切換技術(shù)。簡而言之，其實就是通過一個線程實現(xiàn)代碼塊相互切換執(zhí)行。例如：

def func1():

print(1)

...

print(2)

def func2():

print(3)

...

print(4)

func1()

func2()

上述代碼是普通的函數(shù)定義和執(zhí)行，按流程分別執(zhí)行兩個函數(shù)中的代碼，并先后會輸出：1、2、3、4。但如果介入?yún)f(xié)程技術(shù)那么就可以實現(xiàn)函數(shù)見代碼切換執(zhí)行，最終輸入：1、3、2、4?。

在Python中有多種方式可以實現(xiàn)協(xié)程，例如：

greenlet，是一個第三方模塊，用于實現(xiàn)協(xié)程代碼(Gevent協(xié)程就是基于greenlet實現(xiàn))

yield，生成器，借助生成器的特點也可以實現(xiàn)協(xié)程代碼。

asyncio，在Python3.4中引入的模塊用于編寫協(xié)程代碼。

async & awiat，在Python3.5中引入的兩個關(guān)鍵字，結(jié)合asyncio模塊可以更方便的編寫協(xié)程代碼。

1.1 greenlet

greentlet是一個第三方模塊，需要提前安裝?pip3 install greenlet才能使用。

from greenlet import greenlet

def func1():

print(1) # 第1步：輸出 1

gr2.switch() # 第3步：切換到 func2 函數(shù)

print(2) # 第6步：輸出 2

gr2.switch() # 第7步：切換到 func2 函數(shù)，從上一次執(zhí)行的位置繼續(xù)向后執(zhí)行

def func2():

print(3) # 第4步：輸出 3

gr1.switch() # 第5步：切換到 func1 函數(shù)，從上一次執(zhí)行的位置繼續(xù)向后執(zhí)行

print(4) # 第8步：輸出 4

gr1 = greenlet(func1)

gr2 = greenlet(func2)

gr1.switch() # 第1步：去執(zhí)行 func1 函數(shù)

注意：switch中也可以傳遞參數(shù)用于在切換執(zhí)行時相互傳遞值。

1.2 yield

基于Python的生成器的yield和yield form關(guān)鍵字實現(xiàn)協(xié)程代碼。

def func1():

yield 1

yield from func2()

yield 2

def func2():

yield 3

yield 4

f1 = func1()

for item in f1:

print(item)

注意：yield form關(guān)鍵字是在Python3.3中引入的。

1.3 asyncio

在Python3.4之前官方未提供協(xié)程的類庫，一般大家都是使用greenlet等其他來實現(xiàn)。在Python3.4發(fā)布后官方正式支持協(xié)程，即：asyncio模塊。

import asyncio

@asyncio.coroutine

def func1():

print(1)

yield from asyncio.sleep(2) # 遇到IO耗時操作，自動化切換到tasks中的其他任務

print(2)

@asyncio.coroutine

def func2():

print(3)

yield from asyncio.sleep(2) # 遇到IO耗時操作，自動化切換到tasks中的其他任務

print(4)

tasks = [

asyncio.ensure_future( func1() ),

asyncio.ensure_future( func2() )

]

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(tasks))

注意：基于asyncio模塊實現(xiàn)的協(xié)程比之前的要更厲害，因為他的內(nèi)部還集成了遇到IO耗時操作自動切花的功能。

1.4 async & awit

async & awit 關(guān)鍵字在Python3.5版本中正式引入，基于他編寫的協(xié)程代碼其實就是上一示例的加強版，讓代碼可以更加簡便。

Python3.8之后?@asyncio.coroutine?裝飾器就會被移除，推薦使用async & awit 關(guān)鍵字實現(xiàn)協(xié)程代碼。

import asyncio

async def func1():

print(1)

await asyncio.sleep(2)

print(2)

async def func2():

print(3)

await asyncio.sleep(2)

print(4)

tasks = [

asyncio.ensure_future(func1()),

asyncio.ensure_future(func2())

]

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(tasks))

1.5 小結(jié)

關(guān)于協(xié)程有多種實現(xiàn)方式，目前主流使用是Python官方推薦的asyncio模塊和async&await關(guān)鍵字的方式，例如：在tonado、sanic、fastapi、django3 中均已支持。

接下來，我們也會針對?asyncio模塊?+?async & await?關(guān)鍵字進行更加詳細的講解。

2.協(xié)程的意義

通過學習，我們已經(jīng)了解到協(xié)程可以通過一個線程在多個上下文中進行來回切換執(zhí)行。

但是，協(xié)程來回切換執(zhí)行的意義何在呢？(網(wǎng)上看到很多文章舔協(xié)程，協(xié)程牛逼之處是哪里呢？)

計算型的操作，利用協(xié)程來回切換執(zhí)行，沒有任何意義，來回切換并保存狀態(tài)反倒會降低性能。

IO型的操作，利用協(xié)程在IO等待時間就去切換執(zhí)行其他任務，當IO操作結(jié)束后再自動回調(diào)，那么就會大大節(jié)省資源并提供性能，從而實現(xiàn)異步編程(不等待任務結(jié)束就可以去執(zhí)行其他代碼)。

2.1 爬蟲案例

例如：用代碼實現(xiàn)下載?url_list?中的圖片。

方式一：同步編程實現(xiàn)

"""

下載圖片使用第三方模塊requests，請?zhí)崆鞍惭b：pip3 install requests

"""

import requests

def download_image(url):

print("開始下載:",url)

# 發(fā)送網(wǎng)絡請求，下載圖片

response = requests.get(url)

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'

]

for item in url_list:

download_image(item)

方式二：基于協(xié)程的異步編程實現(xiàn)

"""

下載圖片使用第三方模塊aiohttp，請?zhí)崆鞍惭b：pip3 install aiohttp

"""

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import aiohttp

import asyncio

async def fetch(session, url):

print("發(fā)送請求：", url)

async with session.get(url, verify_ssl=False) as response:

content = await response.content.read()

file_name = url.rsplit('_')[-1]

with open(file_name, mode='wb') as file_object:

file_object.write(content)

async def main():

async with aiohttp.ClientSession() as session:

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 = [asyncio.create_task(fetch(session, url)) for url in url_list]

await asyncio.wait(tasks)

if __name__ == '__main__':

asyncio.run(main())

上述兩種的執(zhí)行對比之后會發(fā)現(xiàn)，基于協(xié)程的異步編程?要比?同步編程的效率高了很多。因為：

同步編程，按照順序逐一排隊執(zhí)行，如果圖片下載時間為2分鐘，那么全部執(zhí)行完則需要6分鐘。

異步編程，幾乎同時發(fā)出了3個下載任務的請求(遇到IO請求自動切換去發(fā)送其他任務請求)，如果圖片下載時間為2分鐘，那么全部執(zhí)行完畢也大概需要2分鐘左右就可以了。

2.2 小結(jié)

協(xié)程一般應用在有IO操作的程序中，因為協(xié)程可以利用IO等待的時間去執(zhí)行一些其他的代碼，從而提升代碼執(zhí)行效率。

生活中不也是這樣的么，假設你是一家制造汽車的老板，員工點擊設備的【開始】按鈕之后，在設備前需等待30分鐘，然后點擊【結(jié)束】按鈕，此時作為老板的你一定希望這個員工在等待的那30分鐘的時間去做點其他的工作。

3.異步編程

基于async?&?await關(guān)鍵字的協(xié)程可以實現(xiàn)異步編程，這也是目前python異步相關(guān)的主流技術(shù)。

想要真正的了解Python中內(nèi)置的異步編程，根據(jù)下文的順序一點點來看。

3.1 事件循環(huán)

事件循環(huán)，可以把他當做是一個while循環(huán)，這個while循環(huán)在周期性的運行并執(zhí)行一些任務，在特定條件下終止循環(huán)。

# 偽代碼

任務列表 = [ 任務1, 任務2, 任務3,... ]

while True:

可執(zhí)行的任務列表，已完成的任務列表 = 去任務列表中檢查所有的任務，將'可執(zhí)行'和'已完成'的任務返回

for 就緒任務 in 已準備就緒的任務列表:

執(zhí)行已就緒的任務

for 已完成的任務 in 已完成的任務列表:

在任務列表中移除已完成的任務

如果任務列表中的任務都已完成，則終止循環(huán)

在編寫程序時候可以通過如下代碼來獲取和創(chuàng)建事件循環(huán)。

import asyncio

loop = asyncio.get_event_loop()

3.2 協(xié)程和異步編程

協(xié)程函數(shù)，定義形式為?async def?的函數(shù)。

協(xié)程對象，調(diào)用?協(xié)程函數(shù)?所返回的對象。

# 定義一個協(xié)程函數(shù)

async def func():

pass

# 調(diào)用協(xié)程函數(shù)，返回一個協(xié)程對象

result = func()

注意：調(diào)用協(xié)程函數(shù)時，函數(shù)內(nèi)部代碼不會執(zhí)行，只是會返回一個協(xié)程對象。

3.2.1 基本應用

程序中，如果想要執(zhí)行協(xié)程函數(shù)的內(nèi)部代碼，需要?事件循環(huán)?和?協(xié)程對象?配合才能實現(xiàn)，如：

import asyncio

async def func():

print("協(xié)程內(nèi)部代碼")

# 調(diào)用協(xié)程函數(shù)，返回一個協(xié)程對象。

result = func()

# 方式一

# loop = asyncio.get_event_loop() # 創(chuàng)建一個事件循環(huán)

# loop.run_until_complete(result) # 將協(xié)程當做任務提交到事件循環(huán)的任務列表中，協(xié)程執(zhí)行完成之后終止。

# 方式二

# 本質(zhì)上方式一是一樣的，內(nèi)部先創(chuàng)建事件循環(huán) 然后執(zhí)行 run_until_complete，一個簡便的寫法。

# asyncio.run 函數(shù)在 Python 3.7 中加入 asyncio 模塊，

asyncio.run(result)

這個過程可以簡單理解為：將協(xié)程當做任務添加到?事件循環(huán)?的任務列表，然后事件循環(huán)檢測列表中的協(xié)程是否已準備就緒(默認可理解為就緒狀態(tài))，如果準備就緒則執(zhí)行其內(nèi)部代碼。

3.2.2 await

await是一個只能在協(xié)程函數(shù)中使用的關(guān)鍵字，用于遇到IO操作時掛起當前協(xié)程(任務)，當前協(xié)程(任務)掛起過程中事件循環(huán)可以去執(zhí)行其他的協(xié)程(任務)，當前協(xié)程IO處理完成時，可以再次切換回來執(zhí)行await之后的代碼。代碼如下：

示例1：

import asyncio

async def func():

print("執(zhí)行協(xié)程函數(shù)內(nèi)部代碼")

# 遇到IO操作掛起當前協(xié)程(任務)，等IO操作完成之后再繼續(xù)往下執(zhí)行。

# 當前協(xié)程掛起時，事件循環(huán)可以去執(zhí)行其他協(xié)程(任務)。

response = await asyncio.sleep(2)

print("IO請求結(jié)束，結(jié)果為：", response)

result = func()

asyncio.run(result)

示例2：

import asyncio

async def others():

print("start")

await asyncio.sleep(2)

print('end')

return '返回值'

async def func():

print("執(zhí)行協(xié)程函數(shù)內(nèi)部代碼")

# 遇到IO操作掛起當前協(xié)程(任務)，等IO操作完成之后再繼續(xù)往下執(zhí)行。當前協(xié)程掛起時，事件循環(huán)可以去執(zhí)行其他協(xié)程(任務)。

response = await others()

print("IO請求結(jié)束，結(jié)果為：", response)

asyncio.run( func() )

示例3：

import asyncio

async def others():

print("start")

await asyncio.sleep(2)

print('end')

return '返回值'

async def func():

print("執(zhí)行協(xié)程函數(shù)內(nèi)部代碼")

response1 = await others()

print("IO請求結(jié)束，結(jié)果為：", response1)

response2 = await others()

print("IO請求結(jié)束，結(jié)果為：", response2)

asyncio.run( func() )

上述的所有示例都只是創(chuàng)建了一個任務，即：事件循環(huán)的任務列表中只有一個任務，所以在IO等待時無法演示切換到其他任務效果。

在程序想要創(chuàng)建多個任務對象，需要使用Task對象來實現(xiàn)。

3.2.3 Task對象

Tasks?are used to schedule coroutines?concurrently.

When a coroutine is wrapped into a?Task?with functions like?asyncio.create_task()?the coroutine is automatically scheduled to run soon。

Tasks用于并發(fā)調(diào)度協(xié)程，通過asyncio.create_task(協(xié)程對象)的方式創(chuàng)建Task對象，這樣可以讓協(xié)程加入事件循環(huán)中等待被調(diào)度執(zhí)行。除了使用?asyncio.create_task()?函數(shù)以外，還可以用低層級的?loop.create_task()?或?ensure_future()?函數(shù)。不建議手動實例化 Task 對象。

本質(zhì)上是將協(xié)程對象封裝成task對象，并將協(xié)程立即加入事件循環(huán)，同時追蹤協(xié)程的狀態(tài)。

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

示例1：

import asyncio

async def func():

print(1)

await asyncio.sleep(2)

print(2)

return "返回值"

async def main():

print("main開始")

# 創(chuàng)建協(xié)程，將協(xié)程封裝到一個Task對象中并立即添加到事件循環(huán)的任務列表中，等待事件循環(huán)去執(zhí)行(默認是就緒狀態(tài))。

task1 = asyncio.create_task(func())

task2 = asyncio.create_task(func())

print("main結(jié)束")

# 當執(zhí)行某協(xié)程遇到IO操作時，會自動化切換執(zhí)行其他任務。

# 此處的await是等待相對應的協(xié)程全都執(zhí)行完畢并獲取結(jié)果

ret1 = await task1

ret2 = await task2

print(ret1, ret2)

asyncio.run(main())

示例2：

import asyncio

async def func():

print(1)

await asyncio.sleep(2)

print(2)

return "返回值"

async def main():

print("main開始")

# 創(chuàng)建協(xié)程，將協(xié)程封裝到Task對象中并添加到事件循環(huán)的任務列表中，等待事件循環(huán)去執(zhí)行(默認是就緒狀態(tài))。

# 在調(diào)用

task_list = [

asyncio.create_task(func(), name="n1"),

asyncio.create_task(func(), name="n2")

]

print("main結(jié)束")

# 當執(zhí)行某協(xié)程遇到IO操作時，會自動化切換執(zhí)行其他任務。

# 此處的await是等待所有協(xié)程執(zhí)行完畢，并將所有協(xié)程的返回值保存到done

# 如果設置了timeout值，則意味著此處最多等待的秒，完成的協(xié)程返回值寫入到done中，未完成則寫到pending中。

done, pending = await asyncio.wait(task_list, timeout=None)

print(done, pending)

asyncio.run(main())

注意：asyncio.wait?源碼內(nèi)部會對列表中的每個協(xié)程執(zhí)行ensure_future從而封裝為Task對象，所以在和wait配合使用時task_list的值為[func(),func()]?也是可以的。

示例3：

import asyncio

async def func():

print("執(zhí)行協(xié)程函數(shù)內(nèi)部代碼")

response = await asyncio.sleep(2)

print("IO請求結(jié)束，結(jié)果為：", response)

coroutine_list = [func(), func()]

# 錯誤：coroutine_list = [ asyncio.create_task(func()), asyncio.create_task(func()) ]

# 此處不能直接 asyncio.create_task，因為將Task立即加入到事件循環(huán)的任務列表，

# 但此時事件循環(huán)還未創(chuàng)建，所以會報錯。

# 使用asyncio.wait將列表封裝為一個協(xié)程，并調(diào)用asyncio.run實現(xiàn)執(zhí)行兩個協(xié)程

# asyncio.wait內(nèi)部會對列表中的每個協(xié)程執(zhí)行ensure_future，封裝為Task對象。

done,pending = asyncio.run( asyncio.wait(coroutine_list) )

3.2.4 asyncio.Future對象

A?Futureis a special?low-level?awaitable object that represents an?eventual result?of an asynchronous operation.

asyncio中的Future對象是一個相對更偏向底層的可對象，通常我們不會直接用到這個對象，而是直接使用Task對象來完成任務的并和狀態(tài)的追蹤。( Task 是 Futrue的子類 )

Future為我們提供了異步編程中的最終結(jié)果的處理(Task類也具備狀態(tài)處理的功能)。

示例1：

async def main():

# 獲取當前事件循環(huán)

loop = asyncio.get_running_loop()

# # 創(chuàng)建一個任務(Future對象)，這個任務什么都不干。

fut = loop.create_future()

# 等待任務最終結(jié)果(Future對象)，沒有結(jié)果則會一直等下去。

await fut

asyncio.run(main())

示例2：

import asyncio

async def set_after(fut):

await asyncio.sleep(2)

fut.set_result("666")

async def main():

# 獲取當前事件循環(huán)

loop = asyncio.get_running_loop()

# 創(chuàng)建一個任務(Future對象)，沒綁定任何行為，則這個任務永遠不知道什么時候結(jié)束。

fut = loop.create_future()

# 創(chuàng)建一個任務(Task對象)，綁定了set_after函數(shù)，函數(shù)內(nèi)部在2s之后，會給fut賦值。

# 即手動設置future任務的最終結(jié)果，那么fut就可以結(jié)束了。

await loop.create_task(set_after(fut))

# 等待 Future對象獲取最終結(jié)果，否則一直等下去

data = await fut

print(data)

asyncio.run(main())

Future對象本身函數(shù)進行綁定，所以想要讓事件循環(huán)獲取Future的結(jié)果，則需要手動設置。而Task對象繼承了Future對象，其實就對Future進行擴展，他可以實現(xiàn)在對應綁定的函數(shù)執(zhí)行完成之后，自動執(zhí)行set_result，從而實現(xiàn)自動結(jié)束。

雖然，平時使用的是Task對象，但對于結(jié)果的處理本質(zhì)是基于Future對象來實現(xiàn)的。

擴展：支持?await 對象語法的對象課成為可等待對象，所以?協(xié)程對象、Task對象、Future對象?都可以被成為可等待對象。

3.2.5 futures.Future對象

在Python的concurrent.futures模塊中也有一個Future對象，這個對象是基于線程池和進程池實現(xiàn)異步操作時使用的對象。

port 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語法等。

官方提示兩對象之間不同：

unlike asyncio Futures,?concurrent.futures.Future?instances cannot be awaited.

Callbacks registered with?asyncio.Future.add_done_callback()?are not called immediately. They are scheduled with?loop.call_soon()?instead.

在Python提供了一個將futures.Future?對象包裝成asyncio.Future對象的函數(shù)?asynic.wrap_future。

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

其實，一般在程序開發(fā)中我們要么統(tǒng)一使用 asycio 的協(xié)程實現(xiàn)異步操作、要么都使用進程池和線程池實現(xiàn)異步操作。但如果?協(xié)程的異步和?進程池/線程池的異步?混搭時，那么就會用到此功能了。

import time

import asyncio

import concurrent.futures

def func1():

# 某個耗時操作

time.sleep(2)

return "SB"

async def main():

loop = asyncio.get_running_loop()

# 1. Run in the default loop's executor ( 默認ThreadPoolExecutor )

# 第一步：內(nèi)部會先調(diào)用 ThreadPoolExecutor 的 submit 方法去線程池中申請一個線程去執(zhí)行func1函數(shù)，并返回一個concurrent.futures.Future對象

# 第二步：調(diào)用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())

應用場景：當項目以協(xié)程式的異步編程開發(fā)時，如果要使用一個第三方模塊，而第三方模塊不支持協(xié)程方式異步編程時，就需要用到這個功能，例如：

import asyncio

import requests

async def download_image(url):

# 發(fā)送網(wǎng)絡請求，下載圖片(遇到網(wǎng)絡下載圖片的IO請求，自動化切換到其他任務)

print("開始下載:", url)

loop = asyncio.get_event_loop()

# requests模塊默認不支持異步操作，所以就使用線程池來配合實現(xiàn)了。

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) )

3.2.6 異步迭代器

什么是異步迭代器

什么是異步可迭代對象？

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():

# 創(chuàng)建異步可迭代對象

async_iter = Reader()

# async for 必須要放在async def函數(shù)內(nèi)，否則語法錯誤。

async for item in async_iter:

print(item)

asyncio.run(func())

異步迭代器其實沒什么太大的作用，只是支持了async for語法而已。

3.2.6 異步上下文管理器

import asyncio

class AsyncContextManager:

def __init__(self):

self.conn = conn

async def do_something(self):

# 異步操作數(shù)據(jù)庫

return 666

async def __aenter__(self):

# 異步鏈接數(shù)據(jù)庫

self.conn = await asyncio.sleep(1)

return self

async def __aexit__(self, exc_type, exc, tb):

# 異步關(guān)閉數(shù)據(jù)庫鏈接

await asyncio.sleep(1)

async def func():

async with AsyncContextManager() as f:

result = await f.do_something()

print(result)

asyncio.run(func())

這個異步的上下文管理器還是比較有用的，平時在開發(fā)過程中打開、處理、關(guān)閉操作時，就可以用這種方式來處理。

3.3 小結(jié)

在程序中只要看到async和await關(guān)鍵字，其內(nèi)部就是基于協(xié)程實現(xiàn)的異步編程，這種異步編程是通過一個線程在IO等待時間去執(zhí)行其他任務，從而實現(xiàn)并發(fā)。

以上就是異步編程的常見操作，內(nèi)容參考官方文檔。

4. uvloop

Python標準庫中提供了asyncio模塊，用于支持基于協(xié)程的異步編程。

uvloop是 asyncio 中的事件循環(huán)的替代方案，替換后可以使得asyncio性能提高。事實上，uvloop要比nodejs、gevent等其他python異步框架至少要快2倍，性能可以比肩Go語言。

安裝uvloop

pip3 install uvloop

在項目中想要使用uvloop替換asyncio的事件循環(huán)也非常簡單，只要在代碼中這么做就行。

import asyncio

import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 編寫asyncio的代碼，與之前寫的代碼一致。

# 內(nèi)部的事件循環(huán)自動化會變?yōu)閡vloop

asyncio.run(...)

注意：知名的asgi uvicorn內(nèi)部就是使用的uvloop的事件循環(huán)。

5.實戰(zhàn)案例

為了更好理解，上述所有示例的IO情況都是以?asyncio.sleep?為例，而真實的項目開發(fā)中會用到很多IO的情況。

5.1 異步Redis

當通過python去操作redis時，鏈接、設置值、獲取值這些都涉及網(wǎng)絡IO請求，使用asycio異步的方式可以在IO等待時去做一些其他任務，從而提升性能。

安裝Python異步操作redis模塊

pip3 install aioredis

示例1：異步操作redis。

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import asyncio

import aioredis

async def execute(address, password):

print("開始執(zhí)行", address)

# 網(wǎng)絡IO操作：創(chuàng)建redis連接

redis = await aioredis.create_redis(address, password=password)

# 網(wǎng)絡IO操作：在redis中設置哈希值car，內(nèi)部在設三個鍵值對，即： redis = { car:{key1:1,key2:2,key3:3}}

await redis.hmset_dict('car', key1=1, key2=2, key3=3)

# 網(wǎng)絡IO操作：去redis中獲取值

result = await redis.hgetall('car', encoding='utf-8')

print(result)

redis.close()

# 網(wǎng)絡IO操作：關(guān)閉redis連接

await redis.wait_closed()

print("結(jié)束", address)

asyncio.run(execute('redis://47.93.4.198:6379', "root!2345"))

示例2：連接多個redis做操作(遇到IO會切換其他任務，提供了性能)。

import asyncio

import aioredis

async def execute(address, password):

print("開始執(zhí)行", address)

# 網(wǎng)絡IO操作：先去連接 47.93.4.197:6379，遇到IO則自動切換任務，去連接47.93.4.198:6379

redis = await aioredis.create_redis_pool(address, password=password)

# 網(wǎng)絡IO操作：遇到IO會自動切換任務

await redis.hmset_dict('car', key1=1, key2=2, key3=3)

# 網(wǎng)絡IO操作：遇到IO會自動切換任務

result = await redis.hgetall('car', encoding='utf-8')

print(result)

redis.close()

# 網(wǎng)絡IO操作：遇到IO會自動切換任務

await redis.wait_closed()

print("結(jié)束", address)

task_list = [

execute('redis://47.93.4.197:6379', "root!2345"),

execute('redis://47.93.4.198:6379', "root!2345")

]

asyncio.run(asyncio.wait(task_list))

5.2 異步MySQL

當通過python去操作MySQL時，連接、執(zhí)行SQL、關(guān)閉都涉及網(wǎng)絡IO請求，使用asycio異步的方式可以在IO等待時去做一些其他任務，從而提升性能。

安裝Python異步操作redis模塊

pip3 install aiomysql

示例1：

import asyncio

import aiomysql

async def execute():

# 網(wǎng)絡IO操作：連接MySQL

conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123', db='mysql', )

# 網(wǎng)絡IO操作：創(chuàng)建CURSOR

cur = await conn.cursor()

# 網(wǎng)絡IO操作：執(zhí)行SQL

await cur.execute("SELECT Host,User FROM user")

# 網(wǎng)絡IO操作：獲取SQL結(jié)果

result = await cur.fetchall()

print(result)

# 網(wǎng)絡IO操作：關(guān)閉鏈接

await cur.close()

conn.close()

asyncio.run(execute())

示例2：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import asyncio

import aiomysql

async def execute(host, password):

print("開始", host)

# 網(wǎng)絡IO操作：先去連接 47.93.40.197，遇到IO則自動切換任務，去連接47.93.40.198:6379

conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')

# 網(wǎng)絡IO操作：遇到IO會自動切換任務

cur = await conn.cursor()

# 網(wǎng)絡IO操作：遇到IO會自動切換任務

await cur.execute("SELECT Host,User FROM user")

# 網(wǎng)絡IO操作：遇到IO會自動切換任務

result = await cur.fetchall()

print(result)

# 網(wǎng)絡IO操作：遇到IO會自動切換任務

await cur.close()

conn.close()

print("結(jié)束", host)

task_list = [

execute('47.93.40.197', "root!2345"),

execute('47.93.40.197', "root!2345")

]

asyncio.run(asyncio.wait(task_list))

5.3 FastAPI框架

FastAPI是一款用于構(gòu)建API的高性能web框架，框架基于Python3.6+的?type hints搭建。

接下里的異步示例以FastAPI和uvicorn來講解(uvicorn是一個支持異步的asgi)。

安裝FastAPI web 框架，

pip3 install fastapi

安裝uvicorn，本質(zhì)上為web提供socket server的支持的asgi(一般支持異步稱asgi、不支持異步稱wsgi)

pip3 install uvicorn

示例：

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import asyncio

import uvicorn

import aioredis

from aioredis import Redis

from fastapi import FastAPI

app = FastAPI()

REDIS_POOL = aioredis.ConnectionsPool('redis://47.193.14.198:6379', password="root123", minsize=1, maxsize=10)

@app.get("/")

def index():

""" 普通操作接口 """

return {"message": "Hello World"}

@app.get("/red")

async def red():

""" 異步操作接口 """

print("請求來了")

await asyncio.sleep(3)

# 連接池獲取一個連接

conn = await REDIS_POOL.acquire()

redis = Redis(conn)

# 設置值

await redis.hmset_dict('car', key1=1, key2=2, key3=3)

# 讀取值

result = await redis.hgetall('car', encoding='utf-8')

print(result)

# 連接歸還連接池

REDIS_POOL.release(conn)

return result

if __name__ == '__main__':

uvicorn.run("luffy:app", host="127.0.0.1", port=5000, log_level="info")

在有多個用戶并發(fā)請求的情況下，異步方式來編寫的接口可以在IO等待過程中去處理其他的請求，提供性能。

例如：同時有兩個用戶并發(fā)來向接口?http://127.0.0.1:5000/red?發(fā)送請求，服務端只有一個線程，同一時刻只有一個請求被處理。異步處理可以提供并發(fā)是因為：當視圖函數(shù)在處理第一個請求時，第二個請求此時是等待被處理的狀態(tài)，當?shù)谝粋€請求遇到IO等待時，會自動切換去接收并處理第二個請求，當遇到IO時自動化切換至其他請求，一旦有請求IO執(zhí)行完畢，則會再次回到指定請求向下繼續(xù)執(zhí)行其功能代碼。

5.4 爬蟲

在編寫爬蟲應用時，需要通過網(wǎng)絡IO去請求目標數(shù)據(jù)，這種情況適合使用異步編程來提升性能，接下來我們使用支持異步編程的aiohttp模塊來實現(xiàn)。

安裝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("得到結(jié)果：", url, len(text))

async def main():

async with aiohttp.ClientSession() as session:

url_list = [

'https://python.org',

'https://www.baidu.com',

'https://www.pythonav.com'

]

tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]

await asyncio.wait(tasks)

if __name__ == '__main__':

asyncio.run(main())

總結(jié)

為了提升性能越來越多的框架都在向異步編程靠攏，例如：sanic、tornado、django3.0、django channels組件等，用更少資源可以做處理更多的事，何樂而不為呢。

內(nèi)容來源于網(wǎng)絡，如有侵權(quán)請聯(lián)系客服刪除

總結(jié)

以上是生活随笔為你收集整理的python异步编程视频_asyncio异步编程【含视频教程】的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。

上一篇：信用卡溢缴款是什么意思
下一篇：华为云设计语言_《好设计，有方法：我们在