Python asyncio 核心原理与高阶应用实战指南

凌晨两点半

发布时间：2026-10-11 | 2人浏览过

原创

asyncio 是 Python 3.4 引入的标准异步 I/O 框架，旨在通过单线程事件循环实现高并发性能。本文从基础语法出发，深入探讨任务管理、并发执行与超时控制等高级技巧，并结合 aiohttp 网络请求、asyncpg 数据库交互及 Web 服务器构建等实际场景，提供完整的代码示例与最佳实践指南。

asyncio 基础与核心机制

asyncio 允许开发者使用 async 和 await 关键字编写异步代码，从而避免阻塞操作。其核心在于事件循环（Event Loop），负责调度和执行协程。

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# 运行异步函数
asyncio.run(main())

高级 asyncio 技巧

任务管理与并发执行

使用 asyncio.create_task() 可以创建任务对象，实现协程的并发执行。通过 asyncio.gather() 可以等待多个任务完成并获取结果。

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():
    # 创建任务
    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())

对于批量并发请求，asyncio.gather() 提供了简洁的接口：

import asyncio

async def fetch_data(id):
    await asyncio.sleep(1)
    return f"Data {id}"

async def main():
    # 并发执行多个任务
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3)
    )
    print(f"Results: {results}")

asyncio.run(main())

超时处理

使用 asyncio.wait_for() 可以为协程设置超时时间，防止长时间阻塞。若超时，将抛出 asyncio.TimeoutError。

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

事件循环操作

虽然 asyncio.run() 是推荐的高级接口，但在某些场景下可能需要直接操作事件循环：

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# 获取事件循环
loop = asyncio.get_event_loop()
# 运行协程
loop.run_until_complete(main())
# 关闭事件循环
loop.close()

实际应用场景

网络请求

结合 aiohttp 库，可以实现高效的并发网络请求：

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"
    ]
    # 并发请求
    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())

文件 I/O

异步文件操作使用 open() 的异步版本，通过 async with 管理上下文：

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

数据库操作

使用 asyncpg 库进行异步数据库交互，注意连接管理：

import asyncio
import asyncpg

async def main():
    # 连接数据库
    conn = await asyncpg.connect(
        host='localhost',
        port=5432,
        user='postgres',
        password='password',
        database='test'
    )
    # 执行查询
    rows = await conn.fetch('SELECT * FROM users')
    for row in rows:
        print(row)
    # 关闭连接
    await conn.close()

asyncio.run(main())

Web 服务器

利用 aiohttp.web 构建异步 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())