Python使用asyncio實現(xiàn)異步操作的示例

更新時間：2025年01月05日 14:32:33 作者：CoderJia_

本文主要介紹了Python使用asyncio實現(xiàn)異步操作的示例,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧

在 Python 中，使用 async 和 await 可以非常高效地處理復雜的異步 I/O 操作。它們的主要目的是簡化異步編程模型，使代碼可讀性更好，并且能夠在 I/O 操作時不阻塞主線程。下面是如何有效地利用它們來實現(xiàn)復雜異步 I/O 操作的指南：

1. 基礎概念

async def：定義一個異步函數(shù)，這樣的函數(shù)在調用時不會立即執(zhí)行，而是返回一個協(xié)程對象。
await：用于等待一個異步操作（如 I/O 操作）的結果，釋放當前函數(shù)持有的 CPU 以便其他協(xié)程能夠執(zhí)行。
asyncio：Python 的標準庫提供了強大的異步 I/O 庫，包含了事件循環(huán)、任務、以及各種異步 I/O 操作的工具。

2. 實現(xiàn)異步 I/O 的步驟

2.1 定義異步函數(shù)

使用 async def 定義異步函數(shù)，可以在函數(shù)內部使用 await 調用異步任務。例如，讀取文件、請求網(wǎng)絡數(shù)據(jù)、或者數(shù)據(jù)庫操作等都可以是異步的。

import asyncio

async def fetch_data():
    print("Fetching data...")
    await asyncio.sleep(2)  # 模擬耗時的 I/O 操作，如數(shù)據(jù)庫查詢或API請求
    return {"data": "sample"}

2.2 使用 await 等待異步操作的完成

通過 await 等待異步任務的完成，可以避免阻塞程序的執(zhí)行。

async def main():
    data = await fetch_data()
    print(data)

# 運行事件循環(huán)
asyncio.run(main())

2.3 并發(fā)執(zhí)行多個任務

通過 asyncio.gather()，你可以并發(fā)地執(zhí)行多個異步任務，而不是順序等待每個任務完成。gather 可以同時啟動多個協(xié)程，并行處理 I/O 操作。

import asyncio


async def task_1():
    await asyncio.sleep(5)
    return "Task 1 finished"


async def task_2():
    await asyncio.sleep(5)
    return "Task 2 finished"


async def main():
    # 計算運行耗時
    start_time = asyncio.get_running_loop().time()
    results = await asyncio.gather(task_1(), task_2())
    print(f"Total time: {asyncio.get_running_loop().time() - start_time}")
    print(results)


asyncio.run(main())

輸出結果是并行執(zhí)行的，雖然 task_1 和 task_2 各需要5秒，但并行總計耗時也是5秒。

2.4 創(chuàng)建并管理任務

asyncio.create_task() 可以將異步函數(shù)封裝成任務，并且不會阻塞當前執(zhí)行。它允許同時運行多個任務，并在它們完成后獲取結果。

import asyncio


async def task_1():
    await asyncio.sleep(2)
    return "Task 1 complete"


async def task_2():
    await asyncio.sleep(1)
    return "Task 2 complete"


async def main():
    # 計算運行的時間
    start_time = asyncio.get_running_loop().time()
    t1 = asyncio.create_task(task_1())
    t2 = asyncio.create_task(task_2())

    await t1  # 等待任務1完成
    await t2  # 等待任務2完成

    print(asyncio.get_running_loop().time() - start_time)
    print(t1.result())
    print(t2.result())


asyncio.run(main())

輸出結果是：

2.5 處理異常

在復雜的異步 I/O 操作中，處理異常非常重要。你可以在 await 和 async 任務中捕獲異常。

async def risky_task():
    await asyncio.sleep(1)
    raise ValueError("An error occurred!")

async def main():
    try:
        await risky_task()
    except ValueError as e:
        print(f"Caught exception: {e}")

asyncio.run(main())

輸出結果是：

2.6 超時控制

異步 I/O 操作中常常需要處理超時情況?？梢酝ㄟ^ asyncio.wait_for() 來實現(xiàn)超時控制。

async def long_task():
    await asyncio.sleep(5)
    return "Task finished"

async def main():
    try:
        result = await asyncio.wait_for(long_task(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

輸出結果是：

3. 處理復雜的異步 I/O 操作

在更復雜的場景中，可能需要同時處理多種類型的 I/O 操作，比如網(wǎng)絡請求、文件讀寫、數(shù)據(jù)庫查詢等。以下是一個例子，它展示了如何通過 asyncio 同時處理不同類型的異步操作。

import asyncio

async def fetch_data_from_api():
    print("Fetching data from API...")
    await asyncio.sleep(3)  # 模擬 API 請求
    return {"api_data": "some api data"}

async def read_from_file():
    print("Reading data from file...")
    await asyncio.sleep(2)  # 模擬文件讀操作
    return "file content"

async def write_to_db(data):
    print(f"Writing {data} to database...")
    await asyncio.sleep(1)  # 模擬數(shù)據(jù)庫寫入操作
    return "DB write success"

async def main():
    # 并發(fā)執(zhí)行 I/O 操作
    api_task = asyncio.create_task(fetch_data_from_api())
    file_task = asyncio.create_task(read_from_file())

    # 等待所有 I/O 操作完成
    api_data, file_content = await asyncio.gather(api_task, file_task)

    # 處理 I/O 操作的結果
    print(f"API Data: {api_data}")
    print(f"File Content: {file_content}")

    # 寫入數(shù)據(jù)庫
    db_result = await write_to_db(api_data)
    print(db_result)

asyncio.run(main())

輸出結果是：

4. 使用 async/await 的性能優(yōu)勢

避免阻塞：傳統(tǒng)的同步 I/O 操作（如文件讀取、網(wǎng)絡請求）會阻塞線程，而 async/await 允許在等待 I/O 操作時執(zhí)行其他任務，極大提高了并發(fā)處理的能力。
降低線程開銷：相比多線程，多協(xié)程（基于 async 的方式）能夠減少線程上下文切換的開銷，在高并發(fā)場景下更加高效。