解锁 Python 异步编程的强大潜力！—— 让你的代码飞速运行！

异步编程：提升应用性能的强大技术

异步编程是一种强大的技术，可以显著提升应用程序的性能。它允许你在不需要手动管理线程或进程的情况下，同时执行多个任务。这种方法特别适用于需要进行 API 请求、文件读取或数据库交互的开发者，而且不会暂停整个应用程序的运行。

在 Python 中，最知名的异步编程库是 asyncio。这个库通过实现事件循环（Event Loop）、协程（Coroutines）和非阻塞 I/O（Non-blocking I/O）来管理程序的执行。如果你希望构建高响应性和高可扩展性的应用程序，强烈建议你深入研究这个库。

同步 vs 异步编程

在同步编程中，任务是按顺序执行的，意味着每个任务必须在前一个任务完成后才能开始。如下图所示，程序会等待一个任务执行完毕后再进行下一个任务。

（同步执行示意图）

相反，异步编程允许多个任务同时运行。当一个任务在等待某个慢操作（如从服务器获取数据）时，程序可以继续执行其他任务。这种**“非阻塞”**方法避免了程序停滞，使可用资源得到更好的利用，从而显著提高执行效率。下图展示了异步编程的工作方式。

（异步执行示意图）

什么是 asyncio？

asyncio 提供了一个框架，使开发者可以使用 async 和 await 语法编写异步代码。它允许程序在不停止其他进程的情况下执行网络请求、文件访问等操作。这样，程序可以在等待某些任务完成时仍然保持响应性。

值得注意的是，asyncio 从 Python 3.3 版本起就默认包含，无需额外安装。

asyncio 的核心组件

事件循环（Event Loop）：管理和调度异步任务和协程的核心机制。
协程（Coroutines）：使用 async def 定义的特殊函数，它们可以使用 await 关键字暂停执行，以便让其他任务运行。
任务（Tasks）：已被调度到事件循环中的协程，可使用 asyncio.create_task() 创建任务。
Future 对象（Futures）：表示可能尚未完成的操作的结果，相当于未来可用的占位符。

示例代码：asyncio 的基本用法

import asyncio

async def calculate_square(number):
    print(f"开始计算 {number} 的平方")
    await asyncio.sleep(1)  # 模拟延迟
    print(f"{number} 的平方计算完成")
    return number ** 2

async def calculate_cube(number):
    print(f"开始计算 {number} 的立方")
    await asyncio.sleep(2)  # 模拟延迟
    print(f"{number} 的立方计算完成")
    return number ** 3

async def main():
    # 创建任务并调度到事件循环中
    task1 = asyncio.create_task(calculate_square(5))  
    task2 = asyncio.create_task(calculate_cube(3))   
    
    # 等待任务完成
    result1 = await task1
    result2 = await task2
   
    # 输出结果
    print(f"平方结果：{result1}")  
    print(f"立方结果：{result2}")   

if __name__ == "__main__":
    # 启动事件循环
    asyncio.run(main())

关键点解析

事件循环：asyncio.run(main()) 启动事件循环，并运行 main() 协程。
协程：calculate_square(number) 和 calculate_cube(number) 是协程，它们使用 await asyncio.sleep() 模拟延迟，而不会阻塞整个程序。
任务：在 main() 中，我们使用 asyncio.create_task() 创建任务，让两个协程并行运行。
Future 对象：result1 和 result2 是 Future 对象，代表 calculate_square() 和 calculate_cube() 任务的最终结果。

运行结果

开始计算 5 的平方
开始计算 3 的立方
5 的平方计算完成
3 的立方计算完成
平方结果：25
立方结果：27

运行多个协程 —— `asyncio.gather()`

asyncio.gather() 允许多个协程并行执行，通常用于批量发送网络请求。让我们用一个示例来模拟多个 HTTP 请求的并发执行。

示例代码

import asyncio

# 模拟从 URL 获取数据的任务
async def fetch_data(URL, delay):
    print(f"开始从 {URL} 获取数据")
    await asyncio.sleep(delay)  # 模拟网络延迟
    return f"从 {URL} 获取数据完成（耗时 {delay} 秒）"

async def main():
    # 并行执行多个获取数据的任务
    results = await asyncio.gather(
        fetch_data("https://openai.com", 2),  
        fetch_data("https://github.com", 5),  
        fetch_data("https://python.org", 7)   
    )
    
    # 输出所有任务的结果
    for result in results:
        print(result)

asyncio.run(main())

关键点解析

fetch_data(URL, delay) 模拟从 URL 获取数据，并设置不同的延迟时间。
asyncio.gather() 让三个 fetch_data() 协程并行执行，而不会一个完成后再启动下一个。
任务并行运行，因此程序不会等待一个任务完成后才开始下一个，极大提升效率。

运行结果

开始从 https://openai.com 获取数据
开始从 https://github.com 获取数据
开始从 https://python.org 获取数据
从 https://openai.com 获取数据完成（耗时 2 秒）
从 https://github.com 获取数据完成（耗时 5 秒）
从 https://python.org 获取数据完成（耗时 7 秒）

处理超时 —— `asyncio.wait_for()`

asyncio.wait_for() 允许你为异步任务设置超时时间，如果任务超时未完成，将引发 asyncio.TimeoutError。

示例代码

import asyncio

async def fetch_data():
    await asyncio.sleep(15)  # 模拟网络延迟
    return "数据获取成功"

async def main():
    try:
        # 设置超时时间为 10 秒
        result = await asyncio.wait_for(fetch_data(), timeout=10) 
        print(result)
    except asyncio.TimeoutError:
        print("数据获取超时")

asyncio.run(main())

如果 fetch_data() 任务执行超过 10 秒，就会触发超时异常 TimeoutError。