1. 为什么需要异步编程在传统的同步编程模型中代码按照顺序逐行执行当遇到I/O密集型操作如网络请求、文件读写时整个程序会被阻塞直到操作完成才能继续执行。这种模式在单线程环境下会造成严重的性能浪费。举个例子假设我们要从三个不同的API端点获取数据import requests def fetch_data_sync(): data1 requests.get(https://api1.example.com).json() # 阻塞1秒 data2 requests.get(https://api2.example.com).json() # 阻塞1秒 data3 requests.get(https://api3.example.com).json() # 阻塞1秒 return [data1, data2, data3]这段代码总耗时至少3秒因为每个请求都是顺序执行的。而使用异步编程这三个请求可以并发执行理论上总耗时只需1秒左右。1.1 事件循环机制Asyncio的核心是事件循环Event Loop它负责调度和执行协程coroutine。事件循环的工作流程如下维护一个任务队列Task Queue从队列中取出就绪的协程执行当协程遇到await表达式时挂起当前协程将控制权交还给事件循环事件循环选择下一个就绪的协程执行当被挂起的协程等待的操作完成时重新加入队列这种机制使得单个线程可以高效地处理大量I/O操作特别适合网络服务器、爬虫等场景。2. Asyncio基础用法2.1 定义协程函数在Python中使用async def语法定义协程函数import asyncio async def say_hello(): print(Hello) await asyncio.sleep(1) # 模拟I/O操作 print(World)协程函数在被调用时不会立即执行而是返回一个协程对象。要运行协程需要将其交给事件循环# Python 3.7推荐方式 asyncio.run(say_hello())2.2 常用API解析2.2.1 asyncio.run()这是Python 3.7引入的高级API用于运行一个协程并管理事件循环async def main(): await asyncio.sleep(1) print(Done) asyncio.run(main())注意asyncio.run()会创建一个新的事件循环并在结束时关闭它。不要在已经运行事件循环的代码中调用它。2.2.2 创建任务使用asyncio.create_task()可以将协程包装为任务Task使其可以并发执行async def task_example(): task1 asyncio.create_task(say_hello()) task2 asyncio.create_task(say_hello()) await task1 await task22.2.3 超时控制asyncio.wait_for()可以为协程设置超时async def fetch_with_timeout(): try: await asyncio.wait_for(slow_operation(), timeout1.0) except asyncio.TimeoutError: print(Operation timed out)3. 实战构建异步网络爬虫3.1 使用aiohttp替代requests传统的requests库是同步的我们需要使用异步HTTP客户端库aiohttpimport aiohttp async def fetch_url(session, url): async with session.get(url) as response: return await response.text() async def crawl(): async with aiohttp.ClientSession() as session: tasks [ fetch_url(session, https://example.com/1), fetch_url(session, https://example.com/2), fetch_url(session, https://example.com/3) ] return await asyncio.gather(*tasks)3.2 控制并发量为了避免同时发起过多请求可以使用信号量Semaphoreasync def bounded_fetch(sem, session, url): async with sem: return await fetch_url(session, url) async def safe_crawl(max_concurrent10): sem asyncio.Semaphore(max_concurrent) async with aiohttp.ClientSession() as session: tasks [ bounded_fetch(sem, session, fhttps://example.com/{i}) for i in range(100) ] return await asyncio.gather(*tasks)3.3 错误处理异步编程中的错误处理需要特别注意async def robust_fetch(session, url): try: async with session.get(url, timeout10) as response: response.raise_for_status() return await response.text() except aiohttp.ClientError as e: print(fRequest failed: {e}) return None4. 高级特性与性能优化4.1 协程与生成器的区别虽然协程和生成器都使用yield语法但它们有本质区别协程是数据的消费者而生成器是数据的生产者协程使用async/await语法生成器使用yield协程可以等待其他协程生成器不能4.2 异步上下文管理器Python 3.7支持异步上下文管理器class AsyncResource: async def __aenter__(self): await self.connect() return self async def __aexit__(self, exc_type, exc, tb): await self.close() async def use_resource(): async with AsyncResource() as resource: await resource.do_something()4.3 异步迭代器实现__aiter__和__anext__方法可以创建异步迭代器class AsyncCounter: def __init__(self, stop): self.current 0 self.stop stop def __aiter__(self): return self async def __anext__(self): if self.current self.stop: raise StopAsyncIteration await asyncio.sleep(0.1) self.current 1 return self.current - 15. 常见问题与调试技巧5.1 协程没有被执行最常见的问题是忘记使用await# 错误示例 async def buggy(): asyncio.sleep(1) # 缺少await协程不会执行 print(This will print immediately) # 正确写法 async def correct(): await asyncio.sleep(1) print(This will print after 1 second)5.2 事件循环已关闭当尝试在asyncio.run()之后访问事件循环时会出现这个问题async def main(): loop asyncio.get_event_loop() # 在asyncio.run()中这是安全的 await asyncio.sleep(1) asyncio.run(main()) loop asyncio.get_event_loop() # 这里会报错事件循环已关闭解决方案是始终在协程内部获取事件循环或者使用asyncio.new_event_loop()创建新循环。5.3 调试异步代码使用asyncio.debug模式可以获取更多调试信息asyncio.run(coro(), debugTrue)或者在代码中设置loop asyncio.get_event_loop() loop.set_debug(True)调试模式下会报告从未被await的协程慢回调执行时间超过100ms资源泄漏警告6. 性能对比与最佳实践6.1 同步vs异步性能测试我们用一个简单的HTTP请求测试来对比import time import aiohttp import requests # 同步版本 def sync_test(): start time.time() for _ in range(10): requests.get(https://httpbin.org/delay/1) print(fSync: {time.time() - start:.2f}s) # 异步版本 async def async_test(): start time.time() async with aiohttp.ClientSession() as session: tasks [ session.get(https://httpbin.org/delay/1) for _ in range(10) ] await asyncio.gather(*tasks) print(fAsync: {time.time() - start:.2f}s)测试结果同步版本约10秒异步版本约1秒6.2 最佳实践总结合理设置并发量不要无限制地创建任务使用信号量控制复用资源如aiohttp的ClientSession应该复用而非每次创建超时设置所有网络操作都应该设置合理的超时错误隔离使用asyncio.shield()保护重要任务不被取消日志记录为每个任务添加唯一标识便于调试避免阻塞操作不要在协程中使用同步I/O或CPU密集型操作7. 与其他异步方案的对比7.1 多线程vs异步特性多线程Asyncio并发模型抢占式多任务协作式多任务上下文切换操作系统控制用户空间控制内存开销较大每个线程MB级较小每个协程KB级适用场景CPU密集型I/O密集型调试难度较难竞态条件相对简单7.2 Asyncio与其他异步框架Twisted老牌异步框架使用回调风格API较复杂Tornado介于回调和协程之间有自己的异步实现Gevent基于greenlet的协程通过monkey patch实现异步Curio/TrioAsyncio的替代方案API设计更一致个人建议新项目优先使用Asyncio它是Python标准库的一部分生态完善。对于已有Twisted/Tornado项目可以继续使用原有框架。8. 实际项目结构建议一个良好的异步项目结构示例project/ ├── main.py # 入口文件 ├── core/ # 核心逻辑 │ ├── __init__.py │ ├── models.py # 数据模型 │ ├── services.py # 业务服务 │ └── utils.py # 工具函数 ├── clients/ # 第三方客户端 │ ├── http.py # HTTP客户端 │ └── database.py # 数据库客户端 └── config.py # 配置文件关键点将I/O操作封装在单独的模块中使用依赖注入而非全局变量为每个服务定义清晰的接口使用类型注解提高代码可维护性9. 异步数据库访问9.1 使用asyncpg访问PostgreSQLimport asyncpg async def query_db(): conn await asyncpg.connect(postgresql://user:passlocalhost/db) try: result await conn.fetch(SELECT * FROM users WHERE id $1, 1) return result finally: await conn.close()9.2 使用aiomysql访问MySQLimport aiomysql async def query_mysql(): conn await aiomysql.connect( hostlocalhost, userroot, password, dbtest ) async with conn.cursor() as cur: await cur.execute(SELECT * FROM users) return await cur.fetchall()9.3 ORM选择推荐使用支持异步的ORMSQLAlchemy 1.4通过asyncpg/aiomysql驱动支持异步Tortoise ORM专为异步设计的ORMGINO基于SQLAlchemy核心的异步ORM10. 测试异步代码10.1 使用pytest-asyncioimport pytest pytest.mark.asyncio async def test_async_code(): result await some_async_function() assert result expected10.2 模拟异步依赖使用unittest.mock的AsyncMockfrom unittest.mock import AsyncMock async def test_with_mock(): mock AsyncMock(return_valuemocked) result await mock() assert result mocked10.3 集成测试技巧使用内存数据库如SQLite加速测试为每个测试创建独立的事件循环使用asyncio.TimeoutError防止测试挂起考虑使用pytest-xdist并行运行测试11. 部署异步应用11.1 使用uvicorn部署ASGI应用对于FastAPI/Sanic等框架uvicorn main:app --workers 4 --host 0.0.0.0 --port 800011.2 性能调优参数--loop uvloop使用更快的uvloop替代默认事件循环--http httptools使用更快的HTTP解析器--limit-concurrency 1000限制最大并发连接数--timeout-keep-alive 5保持连接超时时间11.3 监控与日志推荐工具Prometheus收集性能指标Grafana可视化监控数据Sentry错误跟踪structlog结构化日志记录12. 异步编程设计模式12.1 发布/订阅模式from asyncio import Queue class PubSub: def __init__(self): self.subscribers set() async def publish(self, message): for queue in self.subscribers: await queue.put(message) def subscribe(self): queue Queue() self.subscribers.add(queue) return queue def unsubscribe(self, queue): self.subscribers.remove(queue)12.2 工作队列模式async def worker(queue): while True: task await queue.get() try: await process_task(task) except Exception as e: print(fTask failed: {e}) finally: queue.task_done() async def main(): queue asyncio.Queue(maxsize100) workers [ asyncio.create_task(worker(queue)) for _ in range(5) ] for i in range(1000): await queue.put(ftask-{i}) await queue.join() for w in workers: w.cancel()12.3 断路器模式class CircuitBreaker: def __init__(self, max_failures3, reset_timeout10): self.max_failures max_failures self.reset_timeout reset_timeout self.failures 0 self.state closed async def call(self, coro): if self.state open: raise CircuitOpenError() try: result await coro self.failures 0 return result except Exception: self.failures 1 if self.failures self.max_failures: self.state open asyncio.create_task(self.reset_after_timeout()) raise13. 异步与多进程结合对于CPU密集型任务可以结合多进程和异步import concurrent.futures async def run_in_process(loop, executor, func, *args): return await loop.run_in_executor(executor, func, *args) async def mixed_workload(): loop asyncio.get_running_loop() with concurrent.futures.ProcessPoolExecutor() as pool: cpu_result await run_in_process(loop, pool, cpu_intensive_func, data) io_result await async_io_operation(cpu_result) return io_result14. 异步流处理14.1 使用StreamReader/StreamWriterasync def tcp_echo_client(message): reader, writer await asyncio.open_connection(127.0.0.1, 8888) writer.write(message.encode()) await writer.drain() data await reader.read(100) writer.close() await writer.wait_closed() return data.decode()14.2 处理大文件async def process_large_file(input_path, output_path): async with aiofiles.open(input_path, rb) as infile, \ aiofiles.open(output_path, wb) as outfile: async for line in infile: processed await process_line(line) await outfile.write(processed)15. 异步编程的未来Python的异步生态仍在快速发展中值得关注的趋势结构化并发Trio风格的并发原语可能进入标准库更好的类型支持PEP 612参数规范变量等改进异步生成器改进PEP 525异步生成器的增强与多线程更好集成更简单的线程池交互方式标准库更多异步API如异步文件IO的官方支持在实际项目中采用异步编程时建议从小的、独立的服务开始逐步积累经验。对于复杂的业务系统可以同步和异步组件并存通过队列等方式进行交互。