掌握 Python 异步编程与 asyncio 库 —— 提升高并发编程效率

Python 中的异步编程，尤其是 asyncio 库，是实现高性能应用的强大工具。它可以让代码非阻塞地运行多个任务，非常适合高并发的场景，比如处理大量 I/O 操作或 Web 请求。下面，我们将从基础概念出发，一步步深入，最终带你写出流畅高效的异步代码。

1. 为什么选择异步编程？

异步编程可以显著提升性能，特别是在处理大量 I/O 操作时。传统的同步编程要求每个操作依次完成，意味着在等待一个文件读写或网络响应时，程序会停在那里浪费时间。而异步编程可以在等待时接着处理其他任务，充分利用 CPU，提升整体效率。

比如：假设我们有一个应用需要处理成百上千个 Web 请求，异步编程可以让应用程序一边等待网络响应，一边处理其他请求，而不用卡在那里傻等。这样就可以大大提高系统的吞吐量和响应速度。

2. 同步 vs 异步编程：概念和对比

同步编程：传统上，Python 是同步执行的，也就是按顺序运行每一行代码。如果有个地方需要等待，比如等待网络请求返回，那么后面的代码会一直卡住，直到请求结束。
异步编程：异步编程允许程序在等待的过程中继续运行其他代码。你可以把它想象成一个「会 multitask 的人」，只要有个地方需要等，他会暂时放下这部分，接着做其他事。这个特性在需要并发处理的场景中非常高效。

3. Python 异步编程基础：事件循环、协程和任务

事件循环（Event Loop）是什么？

事件循环是 asyncio 的核心。简单来说，它是一个调度器，负责管理和调度所有的异步任务。所有的协程、任务、I/O 操作等都会提交给事件循环，由它来决定什么先执行，什么后执行。

在 Python 中，你可以使用 asyncio.run() 来启动一个事件循环，并运行你的异步代码。

协程（Coroutine）是什么？

协程是异步代码的基本单元。它有点像普通的函数，但不同的是，它的执行可以暂停（使用 await 关键字），然后在需要的时候继续运行。这样，程序可以切换到其他任务上，不用一直等着协程执行完毕。

在 Python 中，协程通过 async def 定义，比如：

import asyncioasync def my_coroutine():print("这是一个协程")await asyncio.sleep(1)  # 模拟I/O等待print("协程执行完毕")

任务（Task）是什么？

任务是协程的一种包装形式，它是事件循环中的实际执行单元。通过 asyncio.create_task()，可以把协程包装成任务，并提交到事件循环中。

任务和协程的区别是：协程只是一个潜在的任务，只有被事件循环执行时才会变成真正的任务。

4. `asyncio` 的基本用法

我们来看看 asyncio 中几个核心的用法。

启动事件循环并运行协程

import asyncioasync def main():print("启动主协程")await asyncio.sleep(2)print("主协程完成")# 使用 asyncio.run() 启动事件循环并运行主协程
asyncio.run(main())

上面的代码创建了一个事件循环，运行了 main() 协程。await asyncio.sleep(2) 用来模拟一个耗时操作（比如网络请求），程序会等待 2 秒，然后打印完成消息。

创建并发任务

asyncio.create_task() 是让多个任务并发执行的关键。例如，我们同时启动两个任务，看看它们如何并行执行：

import asyncioasync def download_data(name, delay):print(f"{name} 开始下载数据")await asyncio.sleep(delay)print(f"{name} 数据下载完成")async def main():# 创建两个并发任务task1 = asyncio.create_task(download_data("任务1", 2))task2 = asyncio.create_task(download_data("任务2", 1))# 等待所有任务完成await task1await task2asyncio.run(main())

输出顺序会像这样：

任务1 开始下载数据
任务2 开始下载数据
任务2 数据下载完成
任务1 数据下载完成

在这里，任务2比任务1先完成，因为它的 await asyncio.sleep(1) 延迟时间更短。这种并发特性让我们可以高效利用时间。

5. 实战：创建一个异步 Web 爬虫

我们来写个例子：用 asyncio 实现一个异步爬虫。这个小爬虫会访问多个 URL，并行抓取数据。

首先，请确保你已安装 aiohttp和bs4 库：

pip install aiohttp
pip install bs4

接着，我们编写一个可以抓取网页标题的异步爬虫：

import asyncio
import aiohttp
from bs4 import BeautifulSoupasync def fetch_title(url):async with aiohttp.ClientSession() as session:try:async with session.get(url) as response:if response.status == 200:# 读取网页内容html = await response.text()# 解析网页标题soup = BeautifulSoup(html, 'html.parser')title = soup.title.string if soup.title else "无标题"print(f"{url} 的标题是: {title}")else:print(f"抓取 {url} 失败，状态码：{response.status}")except Exception as e:print(f"抓取 {url} 时发生错误: {e}")async def main():urls = ["https://www.taobao.com","https://www.python.org","https://www.jd.com"]# 创建并发任务tasks = [asyncio.create_task(fetch_title(url)) for url in urls]# 等待所有任务完成await asyncio.gather(*tasks)# 运行主协程
asyncio.run(main())

在这个例子中，asyncio.gather(*tasks) 可以将多个任务打包，等待所有任务都完成。这种并发方式让每个 URL 的抓取互不影响。

6. 高级用法：异步上下文管理器、迭代器和队列

异步上下文管理器

异步上下文管理器是一种用于管理异步资源的工具。它的主要作用是在协程中正确地处理需要初始化和清理的操作，例如打开和关闭文件、数据库连接、网络套接字等。与同步上下文管理器类似，异步上下文管理器使用 async with 语法来确保在使用资源时进行必要的管理，并且可以保证无论操作成功与否，资源都会被正确释放。

异步上下文管理器通常用于需要异步初始化和清理的场景，尤其是在高并发或长时间运行的应用中。例如，数据库连接、文件流、网络连接等资源的管理。

如何实现异步上下文管理器

要实现一个异步上下文管理器，需要在类中定义两个特殊方法：__aenter__ 和 __aexit__。这两个方法分别用于在 async with 语句中进入和退出上下文。

__aenter__：当进入 async with 块时调用，它通常用于执行资源的初始化操作，并返回所管理的资源（如数据库连接、文件对象等）。
__aexit__：当 async with 块执行完毕后调用，通常用于执行资源的清理工作，如关闭文件、释放数据库连接等。它接收三个参数，分别是异常类型、异常值和回溯信息，用于处理异常。

示例：自定义异步上下文管理器

我们通过一个简单的例子来展示如何自定义一个异步上下文管理器。这个示例模拟了一个异步资源管理类，该类会打印打开和关闭资源的信息：

import asyncioclass AsyncResource:# 进入上下文时执行的异步操作async def __aenter__(self):print("打开资源")# 可以在此初始化资源，例如连接数据库或打开文件return self# 退出上下文时执行的异步操作async def __aexit__(self, exc_type, exc_val, exc_tb):print("关闭资源")# 可以在此释放资源，例如关闭数据库连接或文件async def main():# 使用异步上下文管理器async with AsyncResource() as resource:print("使用资源")# 在这里可以执行需要的操作，例如使用数据库连接等asyncio.run(main())

代码解释：

__aenter__ 方法：该方法在进入 async with 语句时执行。我们在其中模拟了打开资源的操作。可以在这里实现需要异步处理的资源初始化，如打开异步数据库连接、打开文件流等。__aenter__ 方法的返回值将作为 async with 语句块中的变量（在此例中是 resource）。
__aexit__ 方法：该方法在退出 async with 语句时执行，无论是正常退出还是因为异常退出。我们可以在这里处理资源的清理工作，例如关闭数据库连接、关闭文件流等。如果在 async with 块中发生了异常，__aexit__ 方法会接收异常类型、异常值和回溯信息，可以选择忽略异常或处理它。
async with 语法：在 async with 语句块中，我们使用资源并在块结束后自动执行清理操作。async with 语法保证了资源的安全管理，无论是正常执行完毕，还是发生异常，都会确保 __aexit__ 被调用，从而安全地关闭资源。

运行示例：

打开资源
使用资源
关闭资源

进一步扩展：异步数据库连接

我们可以将这个示例扩展到更复杂的场景，比如模拟一个异步数据库连接。以下是一个更复杂的示例，展示如何使用异步上下文管理器来管理数据库连接：

import asyncioclass AsyncDatabaseConnection:async def __aenter__(self):print("建立数据库连接")# 模拟连接到数据库的异步操作await asyncio.sleep(1)  # 假设数据库连接需要1秒self.connection = "数据库连接对象"return self.connectionasync def __aexit__(self, exc_type, exc_val, exc_tb):print("关闭数据库连接")# 模拟关闭数据库连接的异步操作await asyncio.sleep(1)  # 假设关闭连接需要1秒self.connection = Noneasync def main():# 使用异步上下文管理器管理数据库连接async with AsyncDatabaseConnection() as connection:print("使用数据库连接:", connection)# 在这里执行数据库操作，如查询或插入数据asyncio.run(main())

异步迭代器

异步迭代器适合处理流式数据。比如，当处理来自网络的连续数据时，async for 可以逐个迭代处理。

import asyncioclass AsyncIterator:def __init__(self):self.counter = 0async def __anext__(self):if self.counter < 3:await asyncio.sleep(1)  # 模拟 I/O 延迟self.counter += 1return self.counterelse:raise StopAsyncIterationdef __aiter__(self):return selfasync def main():async for number in AsyncIterator():print(f"收到数据: {number}")asyncio.run(main())

异步队列

asyncio.Queue 适用于协程之间的数据共享。可以将数据从一个任务传递给另一个任务。

import asyncioasync def producer(queue):for i in range(5):await asyncio.sleep(1)await queue.put(f"数据 {i}")print(f"生产数据 {i}")async def consumer(queue):while True:data = await queue.get()print(f"消费 {data}")queue.task_done()async def main():queue = asyncio.Queue()producer_task = asyncio.create_task(producer(queue))consumer_task = asyncio.create_task(consumer(queue))await producer_taskawait queue.join()  # 等待所有任务完成consumer_task.cancel()asyncio.run(main())