Python 異步編程 asyncio簡介及基本用法

更新時間：2025年05月08日 11:54:16 作者：夜闌臥聽風吹雨

asyncio 是 Python 的一個庫,用于編寫并發(fā)代碼,使用協(xié)程、任務和 Futures 來處理 I/O 密集型和高延遲操作,本文給大家介紹Python 異步編程 asyncio簡介及基本用法,感興趣的朋友一起看看吧

1、asyncio是什么

async io（異步輸入輸出）是一種編程模式，它允許程序在等待I/O操作（如文件讀寫、網(wǎng)絡請求等）完成時繼續(xù)執(zhí)行其他任務。通過這種方式，可以更有效地利用CPU資源和提高程序的響應速度，尤其是在處理大量并發(fā)連接或長時間運行的任務時。

asyncio 是 Python 的一個庫，用于編寫并發(fā)代碼，使用協(xié)程、任務和 Futures 來處理 I/O 密集型和高延遲操作

正常的發(fā)送請求，發(fā)送請求后會等待連接建立和服務器響應，這段時間內(nèi)cpu是空閑的，也無法切換到其他任務

asyncio允許當執(zhí)行的任務遇到IO密集型任務時（如發(fā)送請求），cpu空閑，將cpu切換到其他任務上，做其他任務，等請求建立后，可以重新切換回來，繼續(xù)進行操作

IO密集型任務

IO 密集型任務是指那些大部分時間都花費在輸入/輸出（I/O）操作上的任務。這些任務的特點是，CPU 的計算量相對較小，而等待 I/O 操作完成的時間較長。常見的 IO 密集型任務包括文件讀寫、網(wǎng)絡請求、數(shù)據(jù)庫查詢等。

例如連接網(wǎng)絡，發(fā)起連接時需要cpu參與，等待連接過程是不需要cpu參與，這段時間，cpu可以空閑下來，進行其他操作，等待到連接建立成功后需要傳輸數(shù)據(jù)了cpu再進行操作

特征

長時間等待：大量時間花在等待外部設備或服務的響應上。
低 CPU 占用：CPU 在等待 I/O 操作完成時處于空閑狀態(tài)，因為不需要進行復雜的計算。

2、怎么用

1、基本用法

import asyncio
# 定義一個異步函數(shù)
async def my_coroutine():
    print("開始任務")
    # 模擬一個耗時操作
    await asyncio.sleep(1)  # 等待1秒
    print("任務完成")
# 創(chuàng)建事件循環(huán)
loop = asyncio.get_event_loop()
# 運行異步函數(shù)直到完成
loop.run_until_complete(my_coroutine())
# 關(guān)閉事件循環(huán)
loop.close()

定義了一個名為my_coroutine的協(xié)程（coroutine），它會打印一條開始任務，然后暫停執(zhí)行一段時間（模擬I/O操作），最后再打印任務完成。await關(guān)鍵字用于暫停協(xié)程的執(zhí)行，直到等待的操作完成。asyncio.sleep是專門用來模擬異步等待的函數(shù)，在實際應用中，你可能會替換為真正的異步I/O操作，比如網(wǎng)絡請求或文件操作。

如果你想要同時運行多個異步任務，你可以使用asyncio.gather或者asyncio.wait等函數(shù)來管理這些任務。

實際上這樣顯示不出協(xié)程的用途

但是如果有多個任務就可以了

import asyncio
# 定義一個異步函數(shù)
async def my_coroutine1():
    print("開始任務1")
    # 模擬一個耗時操作
    await asyncio.sleep(1)  # 等待1秒
    print("任務完成1")
async def my_coroutine2():
    print("開始任務2")
    # 模擬一個耗時操作
    await asyncio.sleep(1)  # 等待1秒
    print("任務完成2")
async def main():
    results =  await asyncio.gather(my_coroutine1(), my_coroutine2())
    print(results)
if __name__ == '__main__':
    asyncio.run(main())

2、關(guān)鍵字 async

定義：

async關(guān)鍵字用來聲明一個函數(shù)為異步函數(shù)（即協(xié)程）。當一個函數(shù)被定義為異步時，它不會像普通函數(shù)那樣立即執(zhí)行其內(nèi)部代碼，而是返回一個協(xié)程對象。這個協(xié)程對象可以在事件循環(huán)中調(diào)度執(zhí)行。

async def function_name(parameters):
    # 協(xié)程體

使用場景：

當你需要定義一個可能會進行長時間等待的函數(shù)（如網(wǎng)絡請求、文件讀寫等），你可以使用async來標記該函數(shù)，表明它是一個異步操作，并且可以通過await來暫停它的執(zhí)行直到某些條件滿足。

await

定義：await關(guān)鍵字只能出現(xiàn)在由async定義的函數(shù)內(nèi)部，用于暫停當前協(xié)程的執(zhí)行，直到所等待的另一個協(xié)程或異步操作完成。await后面通常跟的是一個可等待對象（如另一個協(xié)程、Future、Task等）。

語法：

await expression

使用場景：

當你在一個異步函數(shù)內(nèi)需要等待另一個異步操作的結(jié)果時，你可以使用await來暫停當前函數(shù)的執(zhí)行，讓出CPU資源給其他任務，直到等待的操作完成。

asyncio.run()：

這是運行頂層協(xié)程的主要方式，適用于Python 3.7+版本。它會創(chuàng)建一個新的事件循環(huán)并運行傳入的協(xié)程，直到協(xié)程完成。完成后，它會關(guān)閉事件循環(huán)并返回協(xié)程的結(jié)果。

asyncio.create_task()：

用于在事件循環(huán)中并發(fā)啟動多個協(xié)程。它接受一個協(xié)程作為參數(shù)，并返回一個Task對象，該對象代表了正在運行的任務。這有助于管理并發(fā)任務，比如同時發(fā)起多個網(wǎng)絡請求。

import asyncio
async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 completed")
async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 completed")
async def main():
    # 創(chuàng)建任務但不等待它們
    task_1 = asyncio.create_task(task1())
    task_2 = asyncio.create_task(task2())
    # 執(zhí)行其他代碼...
    # 等待任務完成
    await task_1
    await task_2
# 運行主函數(shù)
asyncio.run(main())

asyncio.gather()：

如果想要并發(fā)地運行多個協(xié)程并且收集所有結(jié)果，可以使用gather。它可以接收多個協(xié)程作為參數(shù)，并返回一個包含所有協(xié)程返回值的列表。

import asyncio
async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 completed")
    return "Result from Task 1"
async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 completed")
    return "Result from Task 2"
async def main():
    # 使用 gather 同時運行多個任務
    results = await asyncio.gather(task1(), task2())
    print(results)
# 運行主函數(shù)
asyncio.run(main())

asyncio.wait() 可以用來等待一組協(xié)程中的任意數(shù)量完成。你可以指定不同的等待條件，如全部完成、任何一個完成等。

import asyncio
async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 completed")
async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 completed")
async def main():
    # 創(chuàng)建任務列表
    tasks = [task1(), task2()]
    # 等待所有任務完成
    done, pending = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
    for future in done:
        print(future.result())
# 運行主函數(shù)
asyncio.run(main())

如果你想按完成順序處理結(jié)果，而不是按照任務創(chuàng)建的順序，可以使用 asyncio.as_completed()。它返回一個迭代器，在每個協(xié)程完成后產(chǎn)生它的Future對象。

import asyncio
async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 completed")
    return "Result from Task 1"
async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 completed")
    return "Result from Task 2"
async def main():
    # 創(chuàng)建任務列表
    tasks = [task1(), task2()]
    # 按完成順序處理結(jié)果
    for future in asyncio.as_completed(tasks):
        result = await future
        print(result)
# 運行主函數(shù)
asyncio.run(main())