Python异步编程完全指南：从基础到高级应用实战 | Python高级教程

在现代应用开发中，I/O密集型任务（如网络请求、文件操作、数据库查询）往往成为性能瓶颈。异步编程通过非阻塞的方式处理这些任务，让CPU在等待I/O操作时能够执行其他任务，从而大幅提升程序效率。

import asyncio

async def fetch_data(url):
    # 模拟网络请求
    print(f"开始获取 {url}")
    await asyncio.sleep(2)  # 模拟I/O操作
    print(f"完成获取 {url}")
    return f"{url} 的数据"

async def main():
    # 创建多个异步任务
    tasks = [
        fetch_data("https://api.example.com/user"),
        fetch_data("https://api.example.com/products"),
        fetch_data("https://api.example.com/orders")
    ]
    
    # 并发执行所有任务
    results = await asyncio.gather(*tasks)
    print("所有任务完成:", results)

# 运行异步程序
asyncio.run(main())

事件循环工作流程：

1. 执行可运行的协程
2. 处理已完成的任务
3. 执行回调函数
4. 处理I/O事件
5. 重复循环直到所有任务完成

import asyncio
import aiohttp
from bs4 import BeautifulSoup
import time
import json

class AsyncWebCrawler:
    def __init__(self, max_concurrent=10):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []
    
    async def fetch_page(self, session, url):
        async with self.semaphore:
            try:
                async with session.get(url, timeout=30) as response:
                    if response.status == 200:
                        html = await response.text()
                        return await self.parse_page(html, url)
                    else:
                        return {"url": url, "error": f"HTTP {response.status}"}
            except Exception as e:
                return {"url": url, "error": str(e)}
    
    async def parse_page(self, html, url):
        soup = BeautifulSoup(html, 'html.parser')
        title = soup.find('title')
        return {
            "url": url,
            "title": title.text.strip() if title else "No title",
            "timestamp": time.time(),
            "links_count": len(soup.find_all('a'))
        }
    
    async def crawl(self, urls):
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch_page(session, url) for url in urls]
            self.results = await asyncio.gather(*tasks)
    
    def save_results(self, filename):
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)

# 使用示例
async def main():
    urls = [
        "https://httpbin.org/html",
        "https://httpbin.org/json",
        "https://httpbin.org/xml",
        # 可以添加更多URL...
    ] * 3  # 重复URL以演示并发效果
    
    crawler = AsyncWebCrawler(max_concurrent=5)
    
    print("开始异步爬取...")
    start_time = time.time()
    await crawler.crawl(urls)
    end_time = time.time()
    
    print(f"爬取完成！总共处理 {len(urls)} 个URL")
    print(f"耗时: {end_time - start_time:.2f} 秒")
    
    # 保存结果
    crawler.save_results("crawl_results.json")
    print("结果已保存到 crawl_results.json")

# 运行爬虫
asyncio.run(main())

项目亮点：

• 并发控制：使用信号量限制最大并发数
• 错误处理：完善的异常捕获机制
• 性能监控：记录执行时间，评估性能提升
• 数据持久化：结果保存为JSON格式

import aiofiles

class AsyncFileProcessor:
    async def __aenter__(self):
        self.processed_data = []
        return self
    
    async def process_file(self, filename):
        async with aiofiles.open(filename, 'r') as f:
            content = await f.read()
            # 模拟处理过程
            await asyncio.sleep(1)
            processed = content.upper()  # 示例处理
            self.processed_data.append(processed)
            return processed
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f"处理完成，共处理 {len(self.processed_data)} 个文件")

# 使用示例
async def process_files():
    async with AsyncFileProcessor() as processor:
        tasks = [processor.process_file(f"file{i}.txt") for i in range(3)]
        results = await asyncio.gather(*tasks)
        print("处理结果:", results)

async def producer(queue, items):
    for item in items:
        await asyncio.sleep(0.1)  # 模拟生产延迟
        await queue.put(item)
        print(f"生产: {item}")
    await queue.put(None)  # 结束信号

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            queue.put(None)  # 传递给其他消费者
            break
        await asyncio.sleep(0.2)  # 模拟处理延迟
        print(f"消费者 {name} 处理: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    items = range(10)
    
    # 创建生产者和消费者任务
    producer_task = asyncio.create_task(producer(queue, items))
    consumer_tasks = [
        asyncio.create_task(consumer(queue, f"Worker-{i}"))
        for i in range(3)
    ]
    
    await producer_task
    await queue.join()
    
    # 取消消费者任务
    for task in consumer_tasks:
        task.cancel()