Python异步编程完全指南

2026-04-18

字数统计: 1.6k字 | 阅读时长≈ 7分

Python异步编程完全指南：从原理到实践

引言：为什么需要异步编程？

在现代Web应用和数据处理场景中，I/O密集型操作（如网络请求、数据库查询、文件读写）往往成为性能瓶颈。传统的同步编程模型在处理这些操作时，会阻塞整个线程直到操作完成，导致资源利用率低下。

想象一个Web服务器需要同时处理数千个客户端请求：如果使用同步模型，每个请求都会阻塞服务器线程，即使大部分时间都在等待数据库响应或网络传输。异步编程通过非阻塞I/O和事件循环机制，允许单个线程同时处理多个任务，显著提升系统吞吐量。

Python通过asyncio库提供了原生的异步编程支持，自Python 3.5引入async/await语法后，异步编程变得更加直观和易于使用。

技术原理详解

核心概念解析

协程（Coroutine）
协程是异步编程的基本构建块，是一种可以暂停和恢复执行的函数。在Python中，使用async def定义的函数就是协程。

事件循环（Event Loop）
事件循环是异步编程的核心引擎，负责调度和执行协程，处理I/O事件，以及管理回调函数。

Future和Task

Future：表示异步操作的最终结果
Task：是Future的子类，用于包装和管理协程的执行

异步上下文管理器
允许在async with语句中管理异步资源，如数据库连接。

异步迭代器
允许在async for循环中异步遍历数据。

技术术语解释

阻塞I/O：执行I/O操作时，程序会一直等待直到操作完成
非阻塞I/O：I/O操作立即返回，程序可以继续执行其他任务
回调地狱（Callback Hell）：多层嵌套的回调函数导致的代码难以维护的问题
并发（Concurrency）：同时处理多个任务，但不一定同时执行
并行（Parallelism）：同时执行多个任务，需要多核CPU支持

实战代码示例

示例1：基础异步函数

import asyncio
import time

async def fetch_data(delay: int, data_id: int) -> str:
    """模拟异步获取数据"""
    print(f"开始获取数据 {data_id}, 预计耗时 {delay}秒")
    await asyncio.sleep(delay)  # 模拟I/O操作
    print(f"数据 {data_id} 获取完成")
    return f"数据 {data_id} 的内容"

async def main():
    """主协程"""
    print("程序开始")
    
    # 顺序执行
    start_time = time.time()
    result1 = await fetch_data(2, 1)
    result2 = await fetch_data(1, 2)
    print(f"顺序执行结果: {result1}, {result2}")
    print(f"顺序执行耗时: {time.time() - start_time:.2f}秒")
    
    # 并发执行
    print("\n--- 并发执行 ---")
    start_time = time.time()
    task1 = asyncio.create_task(fetch_data(2, 3))
    task2 = asyncio.create_task(fetch_data(1, 4))
    
    results = await asyncio.gather(task1, task2)
    print(f"并发执行结果: {results}")
    print(f"并发执行耗时: {time.time() - start_time:.2f}秒")

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

示例2：异步Web请求

import asyncio
import aiohttp
from typing import List, Dict
import time

class AsyncWebFetcher:
    """异步网页抓取器"""
    
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch_url(self, session: aiohttp.ClientSession, url: str) -> Dict:
        """获取单个URL的内容"""
        async with self.semaphore:  # 限制并发数
            try:
                async with session.get(url, timeout=10) as response:
                    content = await response.text()
                    return {
                        "url": url,
                        "status": response.status,
                        "content_length": len(content),
                        "success": True
                    }
            except Exception as e:
                return {
                    "url": url,
                    "error": str(e),
                    "success": False
                }
    
    async def fetch_multiple_urls(self, urls: List[str]) -> List[Dict]:
        """并发获取多个URL"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch_url(session, url) for url in urls]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return [r for r in results if not isinstance(r, Exception)]

async def main():
    """测试异步网页抓取"""
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/404",
        "https://httpbin.org/status/500",
    ] * 3  # 重复3次以增加任务数量
    
    fetcher = AsyncWebFetcher(max_concurrent=5)
    
    print(f"开始抓取 {len(urls)} 个URL...")
    start_time = time.time()
    
    results = await fetcher.fetch_multiple_urls(urls)
    
    elapsed = time.time() - start_time
    successful = sum(1 for r in results if r.get("success"))
    
    print(f"\n完成！耗时: {elapsed:.2f}秒")
    print(f"成功: {successful}, 失败: {len(results) - successful}")
    print(f"平均每个请求: {elapsed/len(urls):.2f}秒")

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

示例3：异步数据库操作

import asyncio
import asyncpg
from datetime import datetime
from typing import List, Optional

class AsyncDatabase:
    """异步数据库操作类"""
    
    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool: Optional[asyncpg.Pool] = None
    
    async def connect(self, min_size: int = 10, max_size: int = 20):
        """创建连接池"""
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=min_size,
            max_size=max_size,
            command_timeout=60
        )
    
    async def create_table(self):
        """创建示例表"""
        async with self.pool.acquire() as conn:
            await conn.execute('''
                DROP TABLE IF EXISTS users;
                CREATE TABLE users (
                    id SERIAL PRIMARY KEY,
                    username VARCHAR(50) UNIQUE NOT NULL,
                    email VARCHAR(100) UNIQUE NOT NULL,
                    created_at TIMESTAMP DEFAULT NOW(),
                    last_login TIMESTAMP
                )
            ''')
            print("表创建成功")
    
    async def insert_users(self, users: List[dict]):
        """批量插入用户"""
        async with self.pool.acquire() as conn:
            # 使用executemany进行批量插入
            await conn.executemany('''
                INSERT INTO users (username, email) 
                VALUES ($1, $2)
                ON CONFLICT (username) DO NOTHING
            ''', [(u['username'], u['email']) for u in users])
    
    async def get_users(self, limit: int = 100) -> List[dict]:
        """获取用户列表"""
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(
                'SELECT * FROM users ORDER BY created_at DESC LIMIT $1',
                limit
            )
            return [dict(row) for row in rows]
    
    async def update_last_login(self, user_ids: List[int]):
        """批量更新最后登录时间"""
        async with self.pool.acquire() as conn:
            # 使用事务确保一致性
            async with conn.transaction():
                for user_id in user_ids:
                    await conn.execute(
                        'UPDATE users SET last_login = $1 WHERE id = $2',
                        datetime.now(), user_id
                    )
    
    async def close(self):
        """关闭连接池"""
        if self.pool:
            await self.pool.close()

async def main():
    """测试异步数据库操作"""
    # 注意：需要先安装asyncpg并配置PostgreSQL数据库
    # pip install asyncpg
    
    db = AsyncDatabase("postgresql://user:password@localhost/testdb")
    
    try:
        await db.connect()
        await db.create_table()
        
        # 生成测试数据
        users = [
            {"username": f"user{i}", "email": f"user{i}@example.com"}
            for i in range(100)
        ]
        
        # 批量插入
        print("开始插入用户数据...")
        start_time = asyncio.get_event_loop().time()
        await

本文作者： 来的太快的龙卷风
本文链接： https://my-hexo-blog-4i4.pages.dev/2026/04/18/2026-04-18-Python异步编程完全指南-b339f732/
版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！