Python asyncio核心原理與高階應用全解析

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

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

1. asyncio 基礎

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

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

2. 高級 asyncio 技巧

2.1 任務管理

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)建任務
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    # 等待任務完成
    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í)行多個任務
    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:
        # 設置超時
        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()
# 運行協程
loop.run_until_complete(main())
# 關閉事件循環(huán)
loop.close()

3. 實際應用場景

3.1 網絡請求

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 數據庫操作

import asyncio
import asyncpg
async def main():
    # 連接數據庫
    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)
    # 關閉連接
    await conn.close()
asyncio.run(main())

3.4 Web 服務器

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