Python异步编程完全指南：asyncio实战与性能优化技巧

为什么选择异步编程？

• 高性能：避免线程切换开销，提高I/O操作效率
• 资源友好：单线程处理大量并发连接
• 代码简洁：使用async/await语法，代码更易读和维护
• 生态完善：丰富的异步库支持（aiohttp、aiomysql等）

协程（Coroutine）

协程是异步编程的基础构建块，使用async def定义的函数就是协程：

import asyncio

async def simple_coroutine():
    print("开始执行协程")
    await asyncio.sleep(1)
    print("协程执行完成")

# 运行协程
asyncio.run(simple_coroutine())

事件循环（Event Loop）

事件循环是异步编程的核心，负责调度和执行协程任务：

async def main():
    # 创建多个任务
    task1 = asyncio.create_task(some_coroutine())
    task2 = asyncio.create_task(another_coroutine())
    
    # 等待所有任务完成
    await asyncio.gather(task1, task2)

asyncio.run(main())

基本异步函数

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:
        html = await fetch_url(session, 'https://example.com')
        print(f"获取到 {len(html)} 个字符")

asyncio.run(main())

异步上下文管理器

Python 3.7+ 支持异步上下文管理器，用于资源管理：

class AsyncDatabaseConnection:
    async def __aenter__(self):
        self.conn = await connect_to_database()
        return self.conn
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.conn.close()

async def use_database():
    async with AsyncDatabaseConnection() as conn:
        data = await conn.fetch("SELECT * FROM users")
        return data

项目结构设计

async_crawler/
├── crawler.py          # 主爬虫逻辑
├── parser.py           # 页面解析器
├── storage.py          # 数据存储
└── config.py           # 配置管理

核心爬虫实现

import asyncio
import aiohttp
from urllib.parse import urljoin
import async_timeout

class AsyncCrawler:
    def __init__(self, base_url, max_concurrent=10):
        self.base_url = base_url
        self.visited = set()
        self.to_visit = asyncio.Queue()
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        
    async def crawl(self):
        await self.to_visit.put(self.base_url)
        
        workers = [
            asyncio.create_task(self.worker())
            for _ in range(self.max_concurrent)
        ]
        
        await self.to_visit.join()
        
        for worker in workers:
            worker.cancel()
        
        await asyncio.gather(*workers, return_exceptions=True)
    
    async def worker(self):
        while True:
            try:
                url = await self.to_visit.get()
                async with self.semaphore:
                    await self.process_url(url)
            finally:
                self.to_visit.task_done()
    
    async def process_url(self, url):
        if url in self.visited:
            return
        
        self.visited.add(url)
        
        try:
            async with async_timeout.timeout(10):
                async with aiohttp.ClientSession() as session:
                    async with session.get(url) as response:
                        if response.status == 200:
                            html = await response.text()
                            await self.parse_page(url, html)
        except Exception as e:
            print(f"处理 {url} 时出错: {e}")
    
    async def parse_page(self, url, html):
        # 解析HTML并提取新的URL
        print(f"已处理: {url} (长度: {len(html)})")
        
        # 这里添加实际的解析逻辑
        # new_urls = extract_links(html, url)
        # for new_url in new_urls:
        #     if new_url not in self.visited:
        #         await self.to_visit.put(new_url)

# 使用示例
async def main():
    crawler = AsyncCrawler('https://example.com')
    await crawler.crawl()

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

异步生成器

使用异步生成器处理数据流：

async def async_data_stream():
    for i in range(10):
        # 模拟异步数据获取
        await asyncio.sleep(0.1)
        yield i

async def process_stream():
    async for item in async_data_stream():
        print(f"处理: {item}")
        # 进行进一步处理

错误处理与重试机制

import logging
from async_retrying import retry

@retry(attempts=3, delay=1)
async def reliable_api_call(url):
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.json()
    except aiohttp.ClientError as e:
        logging.error(f"API调用失败: {e}")
        raise

性能监控与调试

import time
import asyncio

async def timed_coroutine(func, *args):
    start = time.time()
    result = await func(*args)
    end = time.time()
    print(f"{func.__name__} 执行时间: {end - start:.2f}秒")
    return result

# 使用装饰器进行性能监控
def async_timer(func):
    async def wrapper(*args, **kwargs):
        start = time.time()
        result = await func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 耗时: {end - start:.2f}s")
        return result
    return wrapper

代码组织建议

• 合理划分异步模块和同步模块
• 使用适当的异常处理机制
• 避免在异步函数中执行阻塞操作
• 合理设置并发限制，防止资源耗尽

常见陷阱与解决方案

# 错误示例：在异步函数中使用同步阻塞调用
async def bad_example():
    # 这会阻塞事件循环
    import requests
    response = requests.get('https://example.com')  # 阻塞！
    return response.text

# 正确做法：使用异步HTTP客户端
async def good_example():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://example.com') as response:
            return await response.text()

测试异步代码

import pytest
from your_module import async_function

@pytest.mark.asyncio
async def test_async_function():
    result = await async_function()
    assert result == expected_value