Python异步编程实战第4篇

Python异步编程实战第4篇。本文讲解Python asyncio异步编程的核心概念和实战技巧。Introduction：异步编程是现代高性能Python应用的核心技能，当你的程序需要同时处理大量IO操作（如网络请求、文件读写、数据库查询）时，同步代码会因为等待而效率极低。asyncio是Python3.4引入的标准库，提供了完整的异步编程模型，让你用单线程实现高并发。本文涵盖async/await语法、协程原理、事件循环、并发任务管理、异步上下文管理器、异步迭代器等核心知识点，配完整代码示例，帮助读者快速掌握Python异步编程。

async和await基础：Python用async def定义协程函数，用await等待协程结果。import asyncio async def fetch_data(url): await asyncio.sleep(1) # 模拟网络请求 return f'Data from {url}' async def main(): result = await fetch_data('https://example.com') print(result) asyncio.run(main()) 注意：普通函数不能await，async函数必须通过asyncio.run()或事件循环调用。

事件循环：事件循环是asyncio的核心，它调度和执行所有协程。import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete(main()) finally: loop.close() 或者直接用 asyncio.run() 它会自动管理事件循环。

并发执行多个任务：用asyncio.gather()可以并发执行多个协程。import asyncio async def task(n): await asyncio.sleep(0.5) return n * 2 async def main(): results = await asyncio.gather(*[task(i) for i in range(10)]) print(results) asyncio.run(main()) 10个任务并发执行，总耗时约0.5秒而非5秒。

asyncio.create_task任务调度：asyncio.create_task()将协程提交给事件循环后台执行，而非等待其完成。async def background_task(): while True: await asyncio.sleep(1) print('running') async def main(): task = asyncio.create_task(background_task()) await asyncio.sleep(3) task.cancel() asyncio.run(main()) 这样可以启动后台任务并继续执行其他操作。

异步上下文管理器：用async with处理需要异步初始化和清理的资源。class AsyncDB: async def __aenter__(self): self.conn = await connect() return self.conn async def __aexit__(self, *args): await self.conn.close() async def main(): async with AsyncDB() as conn: result = await conn.query('SELECT * FROM users') print(result) asyncio.run(main())

asyncio.Queue消息队列：asyncio.Queue提供异步安全的队列，支持生产者消费者模式。import asyncio async def producer(q): for i in range(5): await q.put(i) print(f'Produced {i}') async def consumer(q): while True: item = await q.get() print(f'Consumed {item}') q.task_done() async def main(): q = asyncio.Queue() await asyncio.gather(producer(q), consumer(q)) asyncio.run(main())

常见错误处理：协程中的异常需要正确捕获。async def risky_task(): raise ValueError('error') async def main(): try: await risky_task() except ValueError as e: print(f'Caught: {e}') asyncio.run(main()) 异常会在await时抛出，可以用try-except捕获。

以上就是Python异步编程的核心知识点，建议配合实际项目练习。