Python异步编程完全指南

2026-02-10

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

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

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

在现代Web应用和高性能系统中，I/O密集型操作（如网络请求、数据库查询、文件读写）常常成为性能瓶颈。传统的同步编程模型在处理这些操作时，会阻塞整个线程，导致CPU资源浪费和响应延迟。

想象一个简单的场景：你的Web服务器需要同时处理1000个客户端请求，每个请求都需要查询数据库。在同步模型中，服务器必须为每个请求创建一个线程，线程在等待数据库响应时处于空闲状态。这不仅消耗大量内存（每个线程约1MB栈空间），还会因线程切换带来额外开销。

异步编程通过非阻塞I/O和事件循环机制，允许单个线程同时处理多个I/O操作，显著提升系统吞吐量。Python通过asyncio库提供了完整的异步编程支持，让开发者能够编写高效、可扩展的并发代码。

技术原理详解

核心概念解析

1. 协程（Coroutine）
协程是异步编程的基本构建块，它是一种可以暂停执行并在稍后恢复的函数。与线程不同，协程的切换由程序控制，不需要操作系统介入，因此更加轻量级。

import asyncio

# 使用async def定义协程函数
async def fetch_data():
    print("开始获取数据...")
    await asyncio.sleep(2)  # 模拟I/O操作
    print("数据获取完成")
    return {"data": "sample"}

# 协程对象不会立即执行
coro = fetch_data()
print(type(coro))  # <class 'coroutine'>

2. 事件循环（Event Loop）
事件循环是异步编程的核心引擎，负责调度和执行协程，处理I/O事件。它不断检查两个队列：

就绪队列：准备执行的协程
I/O等待队列：等待I/O操作的协程

3. Future和Task

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

异步I/O模型

Python的异步I/O基于操作系统提供的非阻塞I/O接口（如Linux的epoll、macOS的kqueue、Windows的IOCP）。当协程发起I/O请求时，事件循环将其注册到适当的系统接口，然后继续执行其他协程。当I/O操作完成时，系统通知事件循环，相应协程恢复执行。

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        # 同时发起多个HTTP请求
        tasks = [
            fetch_url(session, "https://api.example.com/data1"),
            fetch_url(session, "https://api.example.com/data2"),
            fetch_url(session, "https://api.example.com/data3")
        ]
        results = await asyncio.gather(*tasks)
        return results

实战代码示例

示例1：构建异步Web爬虫

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import time
from urllib.parse import urljoin

class AsyncWebCrawler:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.visited = set()
        
    async def fetch_page(self, session, url):
        """异步获取页面内容"""
        async with self.semaphore:  # 限制并发数
            try:
                async with session.get(url, timeout=10) as response:
                    if response.status == 200:
                        return await response.text()
                    return None
            except Exception as e:
                print(f"Error fetching {url}: {e}")
                return None
    
    async def parse_links(self, html, base_url):
        """解析页面中的链接"""
        if not html:
            return []
        
        soup = BeautifulSoup(html, 'html.parser')
        links = []
        
        for a_tag in soup.find_all('a', href=True):
            href = a_tag['href']
            absolute_url = urljoin(base_url, href)
            if absolute_url.startswith('http'):
                links.append(absolute_url)
        
        return links
    
    async def crawl(self, session, url, depth=2):
        """递归爬取页面"""
        if depth == 0 or url in self.visited:
            return []
        
        self.visited.add(url)
        print(f"Crawling: {url}")
        
        html = await self.fetch_page(session, url)
        if not html:
            return []
        
        # 提取数据（示例：提取标题）
        soup = BeautifulSoup(html, 'html.parser')
        title = soup.title.string if soup.title else "No title"
        print(f"Found: {title}")
        
        if depth > 1:
            links = await self.parse_links(html, url)
            tasks = [self.crawl(session, link, depth-1) for link in links]
            await asyncio.gather(*tasks)
        
        return [{"url": url, "title": title}]
    
    async def run(self, start_urls):
        """启动爬虫"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.crawl(session, url) for url in start_urls]
            results = await asyncio.gather(*tasks)
            return results

# 使用示例
async def main():
    crawler = AsyncWebCrawler(max_concurrent=5)
    start_urls = [
        "https://example.com",
        "https://python.org"
    ]
    
    start_time = time.time()
    results = await crawler.run(start_urls)
    elapsed = time.time() - start_time
    
    print(f"\n爬取完成！耗时: {elapsed:.2f}秒")
    print(f"访问了 {len(crawler.visited)} 个页面")

# 运行爬虫
if __name__ == "__main__":
    asyncio.run(main())

示例2：异步数据库操作

import asyncio
import asyncpg
from datetime import datetime

class AsyncDatabaseManager:
    def __init__(self, dsn):
        self.dsn = dsn
        self.pool = None
    
    async def connect(self):
        """创建数据库连接池"""
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=5,
            max_size=20,
            command_timeout=60
        )
    
    async def create_tables(self):
        """创建示例表"""
        async with self.pool.acquire() as conn:
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id SERIAL PRIMARY KEY,
                    username VARCHAR(50) UNIQUE NOT NULL,
                    email VARCHAR(100) UNIQUE NOT NULL,
                    created_at TIMESTAMP DEFAULT NOW()
                )
            ''')
            
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS orders (
                    id SERIAL PRIMARY KEY,
                    user_id INTEGER REFERENCES users(id),
                    amount DECIMAL(10, 2),
                    status VARCHAR(20),
                    created_at TIMESTAMP DEFAULT NOW()
                )
            ''')
    
    async def batch_insert_users(self, users_data):
        """批量插入用户数据"""
        async with self.pool.acquire() as conn:
            # 使用executemany进行批量插入
            await conn.executemany('''
                INSERT INTO users (username, email) 
                VALUES ($1, $2)
                ON CONFLICT (username) DO NOTHING
            ''', users_data)
    
    async def get_user_with_orders(self, user_id):
        """获取用户及其订单（连接查询）"""
        async with self.pool.acquire() as conn:
            return await conn.fetch('''
                SELECT u.*, o.id as order_id, o.amount, o.status
                FROM users u
                LEFT JOIN orders o ON u.id = o.user_id
                WHERE u.id = $1
                ORDER BY o.created_at DESC
            ''', user_id)
    
    async def concurrent_queries(self, user_ids):
        """并发查询多个用户"""
        tasks = []
        for user_id in user_ids:
            task = self.get_user_with_orders(user_id)
            tasks.append(task)
        
        # 同时执行所有查询
        results = await asyncio.gather(*tasks)
        return results
    
    async def transaction_example(self, user_data, order_data):
        """事务处理示例"""
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                # 插入用户
                user = await conn.fetchrow('''
                    INSERT INTO users (username, email)
                    VALUES ($1, $2)
                    RETURNING id
                ''', user_data['username'], user_data['email'])
                
                # 插入订单
                await conn.execute('''
                    INSERT INTO orders (user_id, amount, status)
                    VALUES ($1, $2, $3)
                ''', user['id'], order_data['amount'], order_data['

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