亚洲乱码中文字幕综合,中国熟女仑乱hd,亚洲精品乱拍国产一区二区三区,一本大道卡一卡二卡三乱码全集资源,又粗又黄又硬又爽的免费视频

Python中asyncio庫實(shí)現(xiàn)異步編程的示例

 更新時(shí)間:2025年04月30日 08:31:16   作者:彬彬俠  
本文主要介紹了Python中asyncio庫實(shí)現(xiàn)異步編程的示例

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

以下是對(duì) asyncio 庫的詳細(xì)說明和常見用法。

1. asyncio 庫的作用

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

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é)程通常通過 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

說明:

  • 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)

說明:

  • 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,以下是常見功能和用法。

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

說明:

  • 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. 與外部庫集成

asyncio 常與異步庫結(jié)合使用,例如:

aiohttp:異步 HTTP 客戶端/服務(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ù)庫操作。

  • aiofiles:異步文件讀寫。

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)行在線程池或進(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)線程池運(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)用場景

  • 網(wǎng)絡(luò)編程:異步爬蟲、Web 服務(wù)器(如 FastAPI、Sanic)。
  • 數(shù)據(jù)庫操作:異步查詢數(shù)據(jù)庫(如 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)

  • 單線程模型:asyncio 在單線程中運(yùn)行,依賴事件循環(huán)調(diào)度,無法利用多核 CPU(需要結(jié)合 multiprocessing)。
  • 避免阻塞代碼:在異步代碼中調(diào)用同步阻塞函數(shù)(如 time.sleep、requests.get)會(huì)阻塞事件循環(huán),需使用異步替代或 run_in_executor。
  • 協(xié)程必須 await:協(xié)程對(duì)象必須通過 await 或任務(wù)調(diào)度運(yùn)行,否則不會(huì)執(zhí)行。
  • 線程安全:事件循環(huá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è)綜合示例,展示異步爬蟲的實(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

說明:

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

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

相關(guān)文章

最新評(píng)論