使用Async IO在Python中進行異步編程的步驟詳解

更新時間：2023年11月02日 09:10:41 作者：我下班楽蟹老板

許多程序員都熟悉編寫順序（同步）代碼,在異步世界中,事件的發(fā)生獨立于主程序流程,異步編程范例有助于并發(fā)執(zhí)行這些任務,并確保您可以克服等待時間并更有效地使用資源,本文給大家介紹了使用Async IO在Python中進行異步編程,需要的朋友可以參考下

想象一下，你有一些獨立的任務，每個任務都需要大量的運行時間才能完成。他們的輸出不相互依賴。所以，您想一次啟動它們。如果這些任務按特定順序執(zhí)行，程序將不得不等待每個任務完成后再開始下一個任務。等待時間會阻塞程序。

異步編程范例有助于并發(fā)執(zhí)行這些任務，并確保您可以克服等待時間并更有效地使用資源。

在 Python 中引入異步

Python 中引入了兩個主要組件：

async io 這是一個 Python 包，允許 API 運行和管理協(xié)程。
async/await 用來定義協(xié)程。

例如要進行 HTTP 調用，請考慮使用 aiohttp ，這是一個 Python 包，允許異步進行 HTTP 調用。在異步代碼中，常用的 requests 庫可能效果不是很好。

同樣，如果您正在使用 Mongo 數(shù)據(jù)庫，而不是依賴同步驅動程序（如 mongo-python ），您必須使用異步驅動程序（如 moto 異步訪問 MongoDB）。

在異步世界中，一切都在事件循環(huán)中運行。這允許您一次運行多個協(xié)程。我們將在本教程中了解協(xié)程是什么。

里面的一切 async def 都是異步代碼，其他一切都是同步的。

編寫異步代碼不像編寫同步代碼那么容易。Python 異步模型基于事件、回調、傳輸、協(xié)議和期貨等概念。

異步如何工作

asyncio 庫提供了兩個關鍵字， async 和 await .

讓我們看一下這個 async hello-world 示例：

import asyncio


async def hello():
    print("Hello World!")
    await asyncio.sleep(1)
    print("Hello again!")


asyncio.run(hello())


# Hello World!
# Hello again!

乍一看你可能認為這是一個同步代碼，因為第二次打印等待 1 秒打印Hello again! 在Hello World!之后。但是這段代碼實際上是異步的。

協(xié)程

任何定義為 async def 的函數(shù)都是像上面那樣的協(xié)程。需要注意的是，調用 hello() 函數(shù)需要在 asyncio.run() 中執(zhí)行，

為了運行協(xié)程，asyncio 提供了三種主要機制：

asyncio.run() 函數(shù)，它是啟動事件循環(huán)并運行異步的主要入口點。

使用 await 協(xié)程的結果并將控制權傳遞給事件循環(huán)。

import asyncio
import time


async def say_something(delay, words):
    print(f"Before {words}")
    await asyncio.sleep(delay)
    print(f"After {words}")
    
async def main():
    print(f"Started: {time.strftime('%X')}") 
    await say_something(1, "Task 1")
    await say_something(2, "Task 2")
    print(f"Finished: {time.strftime( '%X' )}")


asyncio.run(main())


# Started:15:59:52
# Before Task 1
# After Task 1
# Before Task 2
# After Task 2
# Finished:15:59:55

前面的代碼片段仍然等待 say_something() 協(xié)程完成，因此它在 1 秒內執(zhí)行第一個任務，然后在等待 2 秒后執(zhí)行第二個任務。

要讓協(xié)程并發(fā)運行，我們應該創(chuàng)建任務，這是第三種機制。

asyncio.create_task() 用于安排協(xié)程執(zhí)行的函數(shù)。

import asyncio
import time


async def say_something(delay, words):
    print(f"Before {words}")
    await asyncio.sleep(delay)
    print(f"After {words}")
    
async def main():
    print(f"Started: {time.strftime('%X')}")
    task1 = asyncio.create_task(say_something(1,"Task 1")) 
    task2 = asyncio.create_task(say_something(2, "Task 2")) 
    await task1
    await task2
    print(f"Finished: {time.strftime('%X')}")


asyncio.run(main())


# Started:16:07:35
# Before Task 1
# Before Task 2
# After Task 1
# After Task 2
# Finished:16:07:37

上面的代碼現(xiàn)在并發(fā)運行，say_something() 協(xié)程不再等待 say_something() 協(xié)程完成。而是同時運行具有不同參數(shù)的同一個協(xié)程。

發(fā)生的情況如下：

say_something() 協(xié)程從參數(shù)的第一個任務（1 秒和一個字符串Task 1）開始。這個任務叫做 task1。

然后它會暫停協(xié)程的執(zhí)行并等待 1 秒讓 say_something() 協(xié)程在遇到 await 關鍵字時完成。它將控制權返回給事件循環(huán)。

同樣對于第二個任務，它會暫停協(xié)程的執(zhí)行并等待 2 秒讓 say_something() 協(xié)程完成，因為它遇到 await 關鍵字。

task1 控制返回到事件循環(huán)后，事件循環(huán)繼續(xù)執(zhí)行第二個任務 task2 因為 asyncio.sleep() 還沒有完成。

asyncio.create_task() 包裝 say_something() 函數(shù)并使其作為異步任務同時運行協(xié)程。如您所見，上面的代碼片段顯示它的運行速度比以前快了 1 秒。

當 asyncio.create_task() 被調用時，協(xié)程會自動安排在事件循環(huán)中運行。

任務可以幫助您并發(fā)運行多個協(xié)程，但這并不是實現(xiàn)并發(fā)的唯一方法。

使用asyncio.gather() 運行并發(fā)任務

另一種同時運行多個協(xié)程的方法是使用 asyncio.gather() 函數(shù)。此函數(shù)將協(xié)程作為參數(shù)并并發(fā)運行它們。

import asyncio
import time


async def greetings():
    print("Welcome")
    await asyncio.sleep(1)
    print("Goodbye")


async def main():
    start = time.time()
    await asyncio.gather(greetings(), greetings())
    elapsed = time.time() - start
    print(f"{__name__} executed in {elapsed:0.2f} seconds.")


asyncio.run(main())


# Welcome
# Welcome
# Goodbye
# Goodbye
# __main__ executed in 1.00 seconds.

在前面的代碼中，greetings() 協(xié)程被并發(fā)執(zhí)行了兩次。

等待對象

如果一個對象可以與 await 關鍵字一起使用，則該對象稱為可等待對象。可等待對象主要有 3 種類型：coroutines、tasks和futures。

coroutines

import asyncio


async def mult(first, second):
    print("Calculating multiplication...")
    await asyncio.sleep(1)
    mul = first * second
    print(f"{first} multiplied by {second} is {mul}")
    return mul


async def add(first, second):
    print("Calculating sum...")
    await asyncio.sleep(1)
    sum = first + second
    print(f"Sum of {first} and {second} is {sum}")
    return sum


async def main(first, second):
    await mult(first, second)
    await add(first, second)


asyncio.run(main(10, 20))


# Calculating multiplication...
# 10 multiplied by 20 is 200
# Calculating sum...
# Sum of 10 and 20 is 30

在前面的示例中，協(xié)同程序 main() 等待 mult() 和 add() 結束。

假設您在 mult 協(xié)程之前省略了 await 關鍵字。然后您將收到以下錯誤： RuntimeWarning: coroutine 'mult' was never awaited.。

tasks

要安排協(xié)程在事件循環(huán)中運行，我們使用 asyncio.create_task() 函數(shù)。

import asyncio


async def mult(first, second):
    print("Calculating multiplication...")
    await asyncio.sleep(1)
    mul = first * second
    print(f"{first} multiplied by {second} is {mul}")
    return mul


async def add(first, second):
    print("Calculating sum...")
    await asyncio.sleep(1)
    sum = first + second
    print(f"Sum of {first} and {second} is {sum}")
    return sum


async def main(first, second):
    mult_task = asyncio.create_task(mult(first, second))
    add_task = asyncio.create_task(add(first, second))
    await mult_task
    await add_task


asyncio.run(main(10, 20))


# Calculating multiplication...
# Calculating sum...
# 10 multiplied by 20 is 200
# Sum of 10 and 20 is 30

futures

Future 是表示異步計算結果的低級可等待對象。它是通過調用 asyncio.Future() 函數(shù)創(chuàng)建的。

from asyncio import Future


future = Future()
print(future.done())
print(future.cancelled())
future.cancel()
print(future.done())
print(future.cancelled())


# False
# False
# True
# True

超時

用于 asyncio.wait_for(aw, timeout, *) 設置等待對象完成的超時。請注意， aw 這里是可等待的對象。如果要在可等待對象完成時間過長時引發(fā)異常，這將很有用。作為異常 asyncio.TimeoutError。

import asyncio


async def slow_operation():
    await asyncio.sleep(400)
    print("Completed.")


async def main():
    try:
        await asyncio.wait_for(slow_operation(), timeout=1.0)
    except asyncio.TimeoutError:
        print("Timed out!")


asyncio.run(main())


# Timed out!

盡管協(xié)程需要 400 秒才能完成，但 slow_operation() 中的超時 Future 設置為 1 秒。

以上就是使用Async IO在Python中進行異步編程的詳細內容，更多關于Async IO在Python中異步編程的資料請關注腳本之家其它相關文章！

您可能感興趣的文章:

一文詳解凱撒密碼的原理及Python實現(xiàn)
凱撒密碼是古羅馬愷撒大帝用來對軍事情報進行加密的算法，它采用了替換方法對信息中的每一個英文字符循環(huán)替換為字母表序列該字符后面第三個字符。本文主要為大家講解了凱撒密碼的原理及實現(xiàn)，需要的可以參考一下
2022-08-08
Permission denied的解決方法
這篇文章主要介紹了Permission denied的解決方法，希望能給你帶來幫助
2021-08-08
PyCharm中Matplotlib繪圖不能顯示UI效果的問題解決
這篇文章主要介紹了PyCharm中Matplotlib繪圖不能顯示UI效果的問題解決，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2020-03-03
Pandas DataFrame求差集的示例代碼
這篇文章主要介紹了Pandas DataFrame求差集的示例代碼，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2020-12-12
Python裝飾器使用接口測試的步驟
這篇文章主要介紹了Python裝飾器使用接口測試的步驟,本文通過具體示例給大家講解的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
2022-05-05
Python多線程爬蟲實戰(zhàn)_爬取糗事百科段子的實例
下面小編就為大家分享一篇Python多線程爬蟲實戰(zhàn)_爬取糗事百科段子的實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2017-12-12
python中sys.argv函數(shù)精簡概括
本篇文章給大家分享了關于python中sys.argv函數(shù)的相關知識點，有興趣的朋友可以參考學習下。
2018-07-07
Python 二叉樹的層序建立與三種遍歷實現(xiàn)詳解
這篇文章主要介紹了Python 二叉樹的層序建立與三種遍歷實現(xiàn)詳解,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
2019-07-07
python實現(xiàn)音樂下載的統(tǒng)計
這篇文章主要為大家詳細介紹了python實現(xiàn)音樂下載的統(tǒng)計，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2018-06-06
Python?遍歷字典的方法匯總
這篇文章主要介紹了Python?遍歷字典的方法匯總的相關資料,需要的朋友可以參考下
2023-09-09