Python协程高级编程（深入掌握asyncio与异步编程实战）

第一个协程程序

让我们从一个最简单的例子开始：

import asyncioasync def say_hello():    print("Hello")    await asyncio.sleep(1)  # 模拟异步等待    print("World!")# 运行协程asyncio.run(say_hello())

这段代码定义了一个协程函数 say_hello()。注意使用了 async def 声明，内部通过 await asyncio.sleep(1) 暂停执行 1 秒（非阻塞）。最后用 asyncio.run() 启动事件循环并运行协程。

并发执行多个协程

协程真正的威力在于并发处理。比如同时发起多个网络请求或文件读写操作：

import asyncioasync def fetch_data(task_id):    print(f"开始任务 {task_id}")    await asyncio.sleep(2)  # 模拟耗时操作    print(f"完成任务 {task_id}")    return f"结果来自任务 {task_id}"async def main():    # 并发运行多个协程    tasks = [        fetch_data(1),        fetch_data(2),        fetch_data(3)    ]    results = await asyncio.gather(*tasks)    print("所有结果:", results)asyncio.run(main())

这里使用 asyncio.gather() 将多个协程“打包”并发执行。三个任务几乎同时开始，总耗时约 2 秒（而非 6 秒），体现了异步并发的优势。

处理异常与取消任务

在实际开发中，协程可能出错或需要被取消。我们可以使用 try-except 捕获异常，或调用 task.cancel() 主动取消：

import asyncioasync def risky_task():    try:        await asyncio.sleep(1)        raise ValueError("模拟错误！")    except ValueError as e:        print(f"捕获异常: {e}")async def cancellable_task():    try:        await asyncio.sleep(10)    except asyncio.CancelledError:        print("任务已被取消！")        raise  # 重新抛出以确保任务状态正确async def main():    task = asyncio.create_task(cancellable_task())    await asyncio.sleep(0.5)    task.cancel()  # 0.5秒后取消任务    await asyncio.gather(risky_task(), task, return_exceptions=True)asyncio.run(main())

协程 vs 线程 vs 多进程

很多初学者会混淆协程与线程。关键区别如下：

• 协程：单线程内切换，由程序员控制，开销极小，适合 I/O 密集型任务。
• 线程：操作系统调度，有上下文切换开销，受 GIL 限制（CPython），适合混合型任务。
• 多进程：绕过 GIL，真正并行，但内存和通信开销大，适合 CPU 密集型任务。

因此，在 Web 爬虫、API 服务等 I/O 密集场景中，Python协程是最佳选择。

实战：异步 HTTP 请求

使用 aiohttp 库可以高效发起异步 HTTP 请求：

import asyncioimport aiohttpasync def fetch_url(session, url):    async with session.get(url) as response:        return await response.text()async def main():    urls = [        'https://httpbin.org/delay/1',        'https://httpbin.org/delay/2',        'https://httpbin.org/delay/1'    ]    async with aiohttp.ClientSession() as session:        tasks = [fetch_url(session, url) for url in urls]        results = await asyncio.gather(*tasks)        print(f"成功获取 {len(results)} 个响应")asyncio.run(main())

相比同步请求（总耗时约 4 秒），异步版本只需约 2 秒，性能提升显著。

总结

通过本教程，你已掌握了 Python协程 的核心概念、并发执行、异常处理及实际应用场景。记住：异步编程不是万能药，它最适合 I/O 密集型任务。合理使用 asyncio，结合 asyncio教程 中的最佳实践，你就能构建出响应迅速、资源高效的 高性能Python 应用。

下一步建议：尝试用 FastAPI（基于 asyncio）构建 Web API，或用 asyncio 编写分布式爬虫系统，进一步巩固所学知识！