快捷導(dǎo)航

Python asyncio核心原理與高階應(yīng)用全解析

更新時間：2026年05月05日 11:43:12 作者：第一程序員

這篇文章主要介紹了Python asyncio核心原理與高階應(yīng)用,asyncio是Python中強(qiáng)大的異步I/O框架,它允許我們編寫高效的并發(fā)代碼,通過掌握asyncio的高級應(yīng)用,我們可以編寫更加高效、響應(yīng)迅速的應(yīng)用程序,需要的朋友可以參考下

1. asyncio 基礎(chǔ)

asyncio 是 Python 3.4+ 引入的異步 I/O 框架，它允許我們編寫異步代碼，提高程序的并發(fā)性能。

import asyncio
async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")
# 運行異步函數(shù)
asyncio.run(main())

2. 高級 asyncio 技巧

2.1 任務(wù)管理

import asyncio
async def task1():
    await asyncio.sleep(1)
    print("Task 1 completed")
    return "Task 1 result"
async def task2():
    await asyncio.sleep(2)
    print("Task 2 completed")
    return "Task 2 result"
async def main():
    # 創(chuàng)建任務(wù)
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    # 等待任務(wù)完成
    result1 = await task1_obj
    result2 = await task2_obj
    print(f"Results: {result1}, {result2}")
asyncio.run(main())

2.2 并發(fā)執(zhí)行

import asyncio
async def fetch_data(id):
    await asyncio.sleep(1)
    return f"Data {id}"
async def main():
    # 并發(fā)執(zhí)行多個任務(wù)
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3)
    )
    print(f"Results: {results}")
asyncio.run(main())

2.3 超時處理

import asyncio
async def slow_operation():
    await asyncio.sleep(2)
    return "Operation completed"
async def main():
    try:
        # 設(shè)置超時
        result = await asyncio.wait_for(slow_operation(), timeout=1)
        print(f"Result: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out")
asyncio.run(main())

2.4 事件循環(huán)

import asyncio
async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")
# 獲取事件循環(huán)
loop = asyncio.get_event_loop()
# 運行協(xié)程
loop.run_until_complete(main())
# 關(guān)閉事件循環(huán)
loop.close()

3. 實際應(yīng)用場景

3.1 網(wǎng)絡(luò)請求

import asyncio
import aiohttp
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():
    urls = [
        "https://api.github.com",
        "https://api.example.com",
        "https://api.google.com"
    ]
    # 并發(fā)請求
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for url, result in zip(urls, results):
        print(f"URL: {url}, Length: {len(result)}")
asyncio.run(main())

3.2 文件 I/O

import asyncio
async def read_file(filename):
    async with open(filename, 'r') as f:
        content = await f.read()
    return content
async def write_file(filename, content):
    async with open(filename, 'w') as f:
        await f.write(content)
async def main():
    # 讀取文件
    content = await read_file('input.txt')
    print(f"Read content: {content}")
    # 寫入文件
    await write_file('output.txt', content.upper())
    print("File written")
asyncio.run(main())

3.3 數(shù)據(jù)庫操作

import asyncio
import asyncpg
async def main():
    # 連接數(shù)據(jù)庫
    conn = await asyncpg.connect(
        host='localhost',
        port=5432,
        user='postgres',
        password='password',
        database='test'
    )
    # 執(zhí)行查詢
    rows = await conn.fetch('SELECT * FROM users')
    for row in rows:
        print(row)
    # 關(guān)閉連接
    await conn.close()
asyncio.run(main())

3.4 Web 服務(wù)器

import asyncio
from aiohttp import web
async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    return web.Response(text=f"Hello, {name}!")
async def main():
    app = web.Application()
    app.add_routes([
        web.get('/', handle),
        web.get('/{name}', handle)
    ])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    print("Server started on http://localhost:8080")
    await asyncio.Event().wait()
asyncio.run(main())

4. 最佳實踐

使用 async 和 await：使用 async 定義異步函數(shù)，使用 await 等待異步操作完成。
使用 asyncio.create_task：使用 asyncio.create_task 創(chuàng)建任務(wù)，實現(xiàn)并發(fā)執(zhí)行。
使用 asyncio.gather：使用 asyncio.gather 并發(fā)執(zhí)行多個任務(wù)，收集結(jié)果。
使用 asyncio.wait_for：使用 asyncio.wait_for 設(shè)置超時，避免任務(wù)無限等待。
使用 async with：使用 async with 管理異步上下文，確保資源正確釋放。
避免阻塞操作：在異步函數(shù)中避免使用阻塞操作，如 time.sleep()，應(yīng)使用 asyncio.sleep() 替代。
合理使用事件循環(huán)：根據(jù)應(yīng)用場景選擇合適的事件循環(huán)運行方式。