Python 的 asyncio 庫(kù)是標(biāo)準(zhǔn)庫(kù)的一部分，引入于 Python 3.4（通過(guò) PEP 3156），用于實(shí)現(xiàn)異步編程。它提供了一種基于事件循環(huán)（Event Loop）的并發(fā)機(jī)制，適合處理 I/O 密集型任務(wù)（如網(wǎng)絡(luò)請(qǐng)求、文件操作、數(shù)據(jù)庫(kù)查詢(xún)等）。asyncio 通過(guò)協(xié)程（coroutines）、任務(wù)（tasks）和事件循環(huán)，允許程序在等待 I/O 操作時(shí)執(zhí)行其他任務(wù)，從而提高效率。

以下是對(duì) asyncio 庫(kù)的詳細(xì)說(shuō)明和常見(jiàn)用法。

1. asyncio 庫(kù)的作用

• 異步編程：支持非阻塞的并發(fā)執(zhí)行，適合 I/O 密集型任務(wù)。
• 事件循環(huán)：管理協(xié)程和任務(wù)的調(diào)度，協(xié)調(diào)異步操作。
• 高性能：相比線(xiàn)程或進(jìn)程，協(xié)程的開(kāi)銷(xiāo)更低，適合高并發(fā)場(chǎng)景。
• 生態(tài)支持：與許多庫(kù)（如 aiohttp、aiomysql）集成，支持異步網(wǎng)絡(luò)請(qǐng)求、數(shù)據(jù)庫(kù)操作等。

2. 核心概念

• 協(xié)程（Coroutine）：使用 async def 定義的函數(shù)，代表可暫停和恢復(fù)的異步操作。
• 事件循環(huán)（Event Loop）：asyncio 的核心，負(fù)責(zé)調(diào)度協(xié)程和處理 I/O 事件。
• 任務(wù)（Task）：協(xié)程的封裝，用于在事件循環(huán)中并發(fā)運(yùn)行。
• Future：表示尚未完成的操作，協(xié)程通常通過(guò) Future 管理結(jié)果。
• Awaitable：可以被 await 的對(duì)象，包括協(xié)程、任務(wù)和 Future。

3. 基本用法

以下是 asyncio 的基本用法，展示如何定義和運(yùn)行協(xié)程。

3.1 定義和運(yùn)行協(xié)程

import asyncio

# 定義協(xié)程
async def say_hello():
    print("Hello")
    await asyncio.sleep(1)  # 模擬異步 I/O 操作
    print("World")

# 運(yùn)行協(xié)程
async def main():
    await asyncio.gather(say_hello(), say_hello())  # 并發(fā)運(yùn)行多個(gè)協(xié)程

# 執(zhí)行事件循環(huán)
if __name__ == "__main__":
    asyncio.run(main())

輸出：

Hello
Hello
World
World

說(shuō)明：

• async def 定義協(xié)程，await 表示暫停點(diǎn)，允許事件循環(huán)調(diào)度其他任務(wù)。
• asyncio.sleep(1) 模擬異步 I/O（如網(wǎng)絡(luò)請(qǐng)求），不會(huì)阻塞事件循環(huán)。
• asyncio.gather() 并發(fā)運(yùn)行多個(gè)協(xié)程。
• asyncio.run() 是運(yùn)行異步程序的推薦入口，自動(dòng)創(chuàng)建和關(guān)閉事件循環(huán)。

3.2 事件循環(huán)

事件循環(huán)是 asyncio 的核心，負(fù)責(zé)調(diào)度協(xié)程和處理回調(diào)。以下是手動(dòng)操作事件循環(huán)的示例：

import asyncio

async def task():
    print("Task started")
    await asyncio.sleep(1)
    print("Task finished")

loop = asyncio.get_event_loop()  # 獲取事件循環(huán)
try:
    loop.run_until_complete(task())  # 運(yùn)行協(xié)程直到完成
finally:
    loop.close()  # 關(guān)閉事件循環(huán)

說(shuō)明：

• asyncio.get_event_loop() 獲取默認(rèn)事件循環(huán)。
• loop.run_until_complete() 運(yùn)行單個(gè)協(xié)程或 Future。
• 通常推薦使用 asyncio.run()，因?yàn)樗踩易詣?dòng)管理循環(huán)的生命周期。

4. 常用功能

asyncio 提供了豐富的 API，以下是常見(jiàn)功能和用法。

4.1 并發(fā)運(yùn)行多個(gè)協(xié)程

使用 asyncio.gather() 或 asyncio.create_task() 實(shí)現(xiàn)并發(fā)：

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(2)
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(1)
    print("Task 2 finished")

async def main():
    # 使用 gather 并發(fā)運(yùn)行
    await asyncio.gather(task1(), task2())
    # 或者使用 create_task
    t1 = asyncio.create_task(task1())
    t2 = asyncio.create_task(task2())
    await t1
    await t2

asyncio.run(main())

輸出：

Task 1 started
Task 2 started
Task 2 finished
Task 1 finished

說(shuō)明：

• asyncio.gather() 等待所有協(xié)程完成，返回結(jié)果列表。
• asyncio.create_task() 將協(xié)程包裝為任務(wù)，立即調(diào)度運(yùn)行。

4.2 超時(shí)控制

使用 asyncio.wait_for() 為協(xié)程設(shè)置超時(shí)：

import asyncio

async def long_task():
    await asyncio.sleep(5)
    print("Task completed")

async def main():
    try:
        await asyncio.wait_for(long_task(), timeout=2)  # 2 秒超時(shí)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

輸出：

Task timed out

4.3 異步迭代

asyncio 支持異步迭代器和異步上下文管理器：

import asyncio

async def async_generator():
    for i in range(3):
        await asyncio.sleep(1)
        yield i

async def main():
    async for value in async_generator():
        print(f"Received: {value}")

asyncio.run(main())

輸出：

Received: 0
Received: 1
Received: 2

4.4 異步上下文管理器

使用 async with 管理資源：

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def resource():
    print("Resource acquired")
    try:
        yield
    finally:
        print("Resource released")

async def main():
    async with resource():
        print("Using resource")
        await asyncio.sleep(1)

asyncio.run(main())

輸出：

Resource acquired
Using resource
Resource released

5. 與外部庫(kù)集成

asyncio 常與異步庫(kù)結(jié)合使用，例如：

aiohttp：異步 HTTP 客戶(hù)端/服務(wù)器。

import aiohttp
import asyncio

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    html = await fetch_url("https://example.com")
    print(html[:100])

asyncio.run(main())

• aiomysql：異步 MySQL 數(shù)據(jù)庫(kù)操作。

• aiofiles：異步文件讀寫(xiě)。

6. 高級(jí)功能

6.1 任務(wù)取消

可以取消正在運(yùn)行的任務(wù)：

import asyncio

async def long_task():
    try:
        print("Task started")
        await asyncio.sleep(10)
        print("Task finished")
    except asyncio.CancelledError:
        print("Task was cancelled")
        raise

async def main():
    task = asyncio.create_task(long_task())
    await asyncio.sleep(1)
    task.cancel()  # 取消任務(wù)
    try:
        await task
    except asyncio.CancelledError:
        print("Main caught cancellation")

asyncio.run(main())

輸出：

Task started
Task was cancelled
Main caught cancellation

6.2 同步與異步混合

使用 loop.run_in_executor() 將同步代碼（阻塞操作）運(yùn)行在線(xiàn)程池或進(jìn)程池中：

import asyncio
import time

def blocking_task():
    time.sleep(1)  # 模擬阻塞操作
    return "Done"

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_task)  # 在默認(rèn)線(xiàn)程池運(yùn)行
    print(result)

asyncio.run(main())

輸出：

Done

6.3 自定義事件循環(huán)

可以自定義事件循環(huán)策略，例如使用 uvloop（高性能事件循環(huán)）：

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 后續(xù)代碼使用 uvloop 作為事件循環(huán)

安裝 uvloop：

pip install uvloop

7. 實(shí)際應(yīng)用場(chǎng)景

• 網(wǎng)絡(luò)編程：異步爬蟲(chóng)、Web 服務(wù)器（如 FastAPI、Sanic）。
• 數(shù)據(jù)庫(kù)操作：異步查詢(xún)數(shù)據(jù)庫(kù)（如 aiomysql、asyncpg）。
• 實(shí)時(shí)應(yīng)用：聊天 服務(wù)器、WebSocket 應(yīng)用。
• 并發(fā)任務(wù)：批量處理網(wǎng)絡(luò)請(qǐng)求或文件操作。
• 微服務(wù)：異步處理 HTTP 請(qǐng)求或消息隊(duì)列。

8. 注意事項(xiàng)

• 單線(xiàn)程模型：asyncio 在單線(xiàn)程中運(yùn)行，依賴(lài)事件循環(huán)調(diào)度，無(wú)法利用多核 CPU（需要結(jié)合 multiprocessing）。
• 避免阻塞代碼：在異步代碼中調(diào)用同步阻塞函數(shù)（如 time.sleep、requests.get）會(huì)阻塞事件循環(huán)，需使用異步替代或 run_in_executor。
• 協(xié)程必須 await：協(xié)程對(duì)象必須通過(guò) await 或任務(wù)調(diào)度運(yùn)行，否則不會(huì)執(zhí)行。
• 線(xiàn)程安全：事件循環(huán)不是線(xiàn)程安全的，避免在多線(xiàn)程中共享同一循環(huán)。
• 調(diào)試?yán)щy：異步代碼可能因調(diào)度順序?qū)е聫?fù)雜 bug，建議使用日志或調(diào)試工具（如 asyncio.run(debug=True)）。
• Python 版本：
  • Python 3.7+ 引入 asyncio.run() 和上下文管理器改進(jìn)。
  • Python 3.8+ 優(yōu)化了調(diào)試支持。
  • Python 3.11+ 提供任務(wù)組（asyncio.TaskGroup）等新功能。

9. 綜合示例

以下是一個(gè)綜合示例，展示異步爬蟲(chóng)的實(shí)現(xiàn)：

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://example.com",
        "https://python.org",
        "https://github.com"
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"Failed to fetch {url}: {result}")
            else:
                print(f"Fetched {url}: {len(result)} bytes")

if __name__ == "__main__":
    asyncio.run(main())

輸出示例：

Fetched https://example.com: 1256 bytes
Fetched https://python.org: 50342 bytes
Fetched https://github.com: 123456 bytes

說(shuō)明：

• 使用 aiohttp.ClientSession 管理 HTTP 會(huì)話(huà)。
• asyncio.gather 并發(fā)請(qǐng)求多個(gè) URL。
• return_exceptions=True 防止單個(gè)失敗影響其他任務(wù)。

到此這篇關(guān)于Python中asyncio庫(kù)實(shí)現(xiàn)異步編程的示例的文章就介紹到這了,更多相關(guān)Python asyncio異步編程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論