在 Python 中，await 和 asyncio 是異步編程的核心工具，用于高效處理 I/O 密集型任務(wù)（如網(wǎng)絡(luò)請求、文件讀寫、數(shù)據(jù)庫操作等）。它們的核心思想是通過 協(xié)程（Coroutine） 和 事件循環(huán)（Event Loop） 實(shí)現(xiàn)非阻塞并發(fā)，避免線程切換的開銷。

一、核心概念

協(xié)程（Coroutine）
用 async def 定義的函數(shù)，返回一個(gè)協(xié)程對象，可以通過 await 掛起執(zhí)行，讓出控制權(quán)。

async def my_coroutine():
    await asyncio.sleep(1)
    print("Done")

事件循環(huán)（Event Loop）
異步程序的核心調(diào)度器，負(fù)責(zé)執(zhí)行協(xié)程并在 I/O 操作時(shí)切換任務(wù)。

可等待對象（Awaitable）
包括協(xié)程、asyncio.Task 和 asyncio.Future。只有可等待對象才能被 await。

二、使用場景

1. 高并發(fā)網(wǎng)絡(luò)請求
   如爬蟲、API 調(diào)用等需要同時(shí)處理大量連接的場景。
2. Web 服務(wù)器
   如 FastAPI、Sanic 等異步框架處理 HTTP 請求。
3. 數(shù)據(jù)庫操作
   異步驅(qū)動(dòng)（如 asyncpg、aiomysql）避免阻塞主線程。
4. 實(shí)時(shí)通信
   WebSocket、聊天服務(wù)器等需要長連接的場景。

三、基本用法

1. 定義協(xié)程

async def fetch_data(url):
    # 模擬網(wǎng)絡(luò)請求
    await asyncio.sleep(1)
    return f"Data from {url}"

2. 運(yùn)行協(xié)程

async def main():
    result = await fetch_data("https://example.com")
    print(result)
# Python 3.7+ 推薦方式
asyncio.run(main())

3. 并發(fā)執(zhí)行多個(gè)任務(wù)

使用 asyncio.gather() 或 asyncio.create_task()：

async def main():
    # 同時(shí)執(zhí)行多個(gè)協(xié)程
    task1 = asyncio.create_task(fetch_data("url1"))
    task2 = asyncio.create_task(fetch_data("url2"))
    # 等待所有任務(wù)完成
    results = await asyncio.gather(task1, task2)
    print(results)

四、關(guān)鍵 API 詳解

• asyncio.run(coro)
  • 啟動(dòng)事件循環(huán)并運(yùn)行協(xié)程（Python 3.7+）。
• asyncio.create_task(coro)
  • 將協(xié)程包裝為 Task，加入事件循環(huán)并發(fā)執(zhí)行。
• asyncio.gather(*coros)
  • 并發(fā)執(zhí)行多個(gè)協(xié)程，返回結(jié)果列表。
• asyncio.sleep(delay)
  • 非阻塞等待（模擬 I/O 操作）。

五、高級用法

1. 控制并發(fā)量

使用信號量（Semaphore）限制同時(shí)運(yùn)行的任務(wù)數(shù)：

async def limited_fetch(url, semaphore):
    async with semaphore:
        return await fetch_data(url)
async def main():
    semaphore = asyncio.Semaphore(5)  # 最大并發(fā)5
    tasks = [limited_fetch(url, semaphore) for url in urls]
    await asyncio.gather(*tasks)

2. 超時(shí)控制

async def fetch_with_timeout():
    try:
        async with asyncio.timeout(3):  # Python 3.11+
            await fetch_data("slow_url")
    except TimeoutError:
        print("Timeout!")

3. 回調(diào)與 Future

async def main():
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    def callback():
        future.set_result("Done")
    loop.call_soon(callback)
    result = await future
    print(result)

六、常見錯(cuò)誤

忘記 await
協(xié)程不會(huì)被自動(dòng)執(zhí)行：

async def main():
    fetch_data("url")  # 錯(cuò)誤！沒有 await

阻塞主線程
在協(xié)程中調(diào)用同步代碼（如 time.sleep()）會(huì)阻塞事件循環(huán)：

async def bad_example():
    time.sleep(1)  # 錯(cuò)誤！應(yīng)使用 await asyncio.sleep(1)

濫用并發(fā)
異步不適合 CPU 密集型任務(wù)，此時(shí)應(yīng)使用多進(jìn)程。

七、完整示例

import asyncio
async def download(url, delay):
    print(f"Start downloading {url}")
    await asyncio.sleep(delay)
    print(f"Finished {url}")
    return url
async def main():
    urls = [
        ("https://site1.com", 1),
        ("https://site2.com", 2),
        ("https://site3.com", 3),
    ]
    tasks = [asyncio.create_task(download(url, delay)) for url, delay in urls]
    results = await asyncio.gather(*tasks)
    print("All done:", results)
if __name__ == "__main__":
    asyncio.run(main())

輸出：

Start downloading https://site1.com
Start downloading https://site2.com
Start downloading https://site3.com
Finished https://site1.com
Finished https://site2.com
Finished https://site3.com
All done: ['https://site1.com', 'https://site2.com', 'https://site3.com']

八、總結(jié)

• 適用場景：I/O 密集型任務(wù)，如網(wǎng)絡(luò)、文件、數(shù)據(jù)庫操作。
• 關(guān)鍵點(diǎn)：
  • 使用 async def 定義協(xié)程，用 await 掛起阻塞操作。
  • 通過 asyncio.create_task() 和 asyncio.gather() 實(shí)現(xiàn)并發(fā)。
  • 避免在協(xié)程中調(diào)用阻塞同步代碼。

通過合理使用 asyncio，可以在單線程內(nèi)高效處理成千上萬的并發(fā)連接。

到此這篇關(guān)于Python異步編程之a(chǎn)wait與asyncio基本用法詳解的文章就介紹到這了,更多相關(guān)Python await與asyncio內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論